test(ecstore): distinguish seed initialization from repair

The namespace owner tests decided that ownership had ended when the Weak probe stopped upgrading or when the mutation lease could be reacquired. Both signals fire before the owner guard's Drop decrements the pending counter: Arc releases its strong count before running Drop, and the lease drops its locks before its owner field. The rio-v2 lane hit that window in undo_fresh_version_keeps_physical_namespace_owner_after_timeout.

Extend every drain wait to also require namespace_commits_pending() to be false, so the assertions observe the completed release instead of racing it.

.github/workflows/ci.yml

@@ -582,13 +582,59 @@ jobs:
           install-build-packaging-tools: 'false'
 
       - name: Build debug binary
-        run: cargo build -p rustfs --bins --features e2e-test-hooks
+        run: |
+          python3 - <<'PYBUILD'
+          import hashlib
+          import json
+          import os
+          import pathlib
+          import subprocess
+
+          def git(*args):
+              return subprocess.check_output(["git", *args], text=True).strip()
+
+          def sha256(path):
+              digest = hashlib.sha256()
+              with pathlib.Path(path).open("rb") as source:
+                  for chunk in iter(lambda: source.read(1024 * 1024), b""):
+                      digest.update(chunk)
+              return digest.hexdigest()
+
+          argv = ["cargo", "build", "-p", "rustfs", "--bins", "--features", "e2e-test-hooks"]
+          commit, tree = git("rev-parse", "HEAD"), git("rev-parse", "HEAD^{tree}")
+          clean_before = not git("status", "--porcelain", "--untracked-files=normal")
+          if not clean_before:
+              raise SystemExit("hooks binary requires a clean build checkout")
+          lock_sha256 = sha256("Cargo.lock")
+          lock_git_blob = git("hash-object", "Cargo.lock")
+          rustc = subprocess.check_output(["rustc", "-vV"], text=True)
+          host = next(line.removeprefix("host: ") for line in rustc.splitlines() if line.startswith("host: "))
+          if os.environ.get("CARGO_BUILD_TARGET") or pathlib.Path(os.environ.get("CARGO_TARGET_DIR", "target")).resolve() != pathlib.Path("target").resolve():
+              raise SystemExit("this artifact requires the native target/debug output")
+          subprocess.run(argv, check=True)
+          clean_after = not git("status", "--porcelain", "--untracked-files=normal")
+          if not clean_after or commit != git("rev-parse", "HEAD") or tree != git("rev-parse", "HEAD^{tree}") or lock_sha256 != sha256("Cargo.lock"):
+              raise SystemExit("hooks binary source changed while building")
+          manifest = {
+              "schema": 1, "commit": commit, "tree": tree,
+              "clean_before": clean_before, "clean_after": clean_after,
+              "lock_sha256": lock_sha256, "lock_git_blob": lock_git_blob,
+              "argv": argv, "profile": "debug", "target": host,
+              "features": ["e2e-test-hooks"],
+              "rustc_verbose": rustc,
+              "build_flags": {key: os.environ[key] for key in ("RUSTFLAGS", "CARGO_ENCODED_RUSTFLAGS", "CARGO_BUILD_TARGET", "CARGO_TARGET_DIR", "RUSTUP_TOOLCHAIN") if key in os.environ},
+              "binary_sha256": sha256("target/debug/rustfs"),
+          }
+          pathlib.Path("target/debug/rustfs.e2e-startup-cas-build.json").write_text(json.dumps(manifest, indent=2) + "\n")
+          PYBUILD
 
       - name: Upload debug binary
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
           name: rustfs-debug-binary
-          path: target/debug/rustfs
+          path: |
+            target/debug/rustfs
+            target/debug/rustfs.e2e-startup-cas-build.json
           if-no-files-found: error
           retention-days: 1
 
@@ -906,6 +952,36 @@ jobs:
       - name: Make binary executable
         run: chmod +x ./target/debug/rustfs
 
+      - name: Preserve startup CAS binary input
+        env:
+          STARTUP_CAS_INPUT: ${{ runner.temp }}/rustfs-startup-cas-input
+        run: |
+          python3 - <<'PYINPUT'
+          import hashlib
+          import json
+          import os
+          import pathlib
+          import shutil
+          import subprocess
+
+          source = pathlib.Path("target/debug/rustfs")
+          manifest_path = source.with_name("rustfs.e2e-startup-cas-build.json")
+          manifest = json.loads(manifest_path.read_text())
+          target = pathlib.Path(os.environ["STARTUP_CAS_INPUT"])
+          target.mkdir(parents=True, exist_ok=True)
+          binary = target / "rustfs"
+          shutil.copy2(source, binary)
+          digest = hashlib.sha256()
+          with binary.open("rb") as stream:
+              for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+                  digest.update(chunk)
+          commit = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()
+          if manifest["binary_sha256"] != digest.hexdigest() or manifest["commit"] != commit:
+              raise SystemExit("downloaded hooks binary identity mismatch")
+          shutil.copy2(manifest_path, target / manifest_path.name)
+          binary.chmod(0o755)
+          PYINPUT
+
       - name: Verify e2e full membership
         env:
           NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-full-list.json
@@ -918,6 +994,10 @@ jobs:
       # extend that filter, never add ad-hoc e2e jobs here. Reuses the downloaded
       # debug binary; each test spawns its own rustfs server on a random port.
       - name: Run e2e full suite
+        env:
+          RUSTFS_E2E_STARTUP_CAS_BINARY: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs
+          RUSTFS_E2E_STARTUP_CAS_BUILD_MANIFEST: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs.e2e-startup-cas-build.json
+          RUSTFS_E2E_STARTUP_CAS_ARTIFACT_DIR: ${{ runner.temp }}/rustfs-startup-cas-evidence
         run: cargo nextest run --profile e2e-full -p e2e_test
 
       - name: Upload junit
@@ -930,6 +1010,17 @@ jobs:
             ${{ runner.temp }}/rustfs-e2e-full-list.json
           retention-days: 7
 
+      - name: Upload startup CAS evidence
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: fresh-startup-cas-evidence-${{ github.run_number }}
+          path: |
+            ${{ runner.temp }}/rustfs-startup-cas-evidence
+            ${{ runner.temp }}/rustfs-startup-cas-input/rustfs.e2e-startup-cas-build.json
+          if-no-files-found: warn
+          retention-days: 7
+
   e2e-tests-rio-v2:
     name: End-to-End Tests (rio-v2)
     # Inherits the schedule/dispatch-only gate through needs: on every other

Cargo.lock

@@ -4057,6 +4057,7 @@ dependencies = [
  "sha1 0.11.0",
  "sha2 0.11.0",
  "suppaftp",
+ "tempfile",
  "time",
  "tokio",
  "tokio-stream",

crates/e2e_test/Cargo.toml

@@ -144,3 +144,6 @@ russh = { workspace = true, features = ["serde"] }
 russh-sftp = { workspace = true }
 zip.workspace = true
 clap = { workspace = true, features = ["derive", "env"] }
+
+[dev-dependencies]
+tempfile.workspace = true

crates/ecstore/Cargo.toml

@@ -118,6 +118,8 @@ hotpath-cpu = [
 # injection, xl.meta transition assertions) via `api::tier::test_util`.
 # Enable only from `[dev-dependencies]` (rustfs/backlog#1148 ilm-6).
 test-util = []
+# Observes real startup CAS only in the dedicated E2E binary.
+e2e-test-hooks = []
 
 [dependencies]
 hotpath.workspace = true

crates/ecstore/src/api/mod.rs

@@ -368,6 +368,8 @@ pub mod data_usage {
 pub mod disk {
     pub use crate::disk::disk_store::get_object_disk_read_timeout;
     pub use crate::disk::local::ScanGuard;
+    #[cfg(all(feature = "test-util", not(windows)))]
+    pub use crate::disk::os::{LocalPublicationPause, LocalPublicationStage};
     pub use crate::disk::{
         BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
         CheckPartsResp, ConditionalFileUpdate, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
@@ -544,8 +546,8 @@ pub mod storage {
     pub use crate::core::pools::HealLifecycleExpiryContext;
     pub use crate::store::HealWalkVersion;
     pub use crate::store::{
-        ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, all_local_disk_path,
-        find_local_disk_by_ref, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients,
+        BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
+        all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients,
         prewarm_local_disk_id_map, prewarm_local_disk_id_map_with_instance_ctx,
     };
 }

crates/ecstore/src/core/pools.rs

@@ -5108,7 +5108,49 @@ async fn read_pool_meta_replicas<S>(pools: Vec<Arc<S>>, no_lock: bool) -> Vec<Po
 where
     S: EcstoreObjectIO,
 {
-    join_all(pools.into_iter().map(|pool| read_pool_meta_replica(pool, no_lock))).await
+    let reads = join_all(pools.into_iter().map(|pool| read_pool_meta_replica(pool, no_lock))).await;
+    #[cfg(feature = "e2e-test-hooks")]
+    if STARTUP_CAS_OBSERVATION.try_with(|_| ()).is_ok() {
+        let batch = uuid::Uuid::new_v4();
+        for (pool, read) in reads.iter().enumerate() {
+            let mut observation = serde_json::json!({
+                "kind": "replica-read", "object": POOL_META_NAME, "batch": batch, "pool": pool,
+                "cas": match &read.cas {
+                    PoolMetaCasToken::Missing => "missing",
+                    PoolMetaCasToken::Existing(_) => "existing",
+                    PoolMetaCasToken::Unsafe => "unsafe",
+                },
+                "etag": match &read.cas { PoolMetaCasToken::Existing(etag) => Some(etag), _ => None },
+            });
+            match &read.replica {
+                PoolMetaReplica::Valid {
+                    raw,
+                    canonical,
+                    meta,
+                    revision,
+                    committed,
+                    ..
+                } => {
+                    observation["state"] = serde_json::json!("valid");
+                    observation["committed"] = serde_json::json!(committed);
+                    observation["version"] = serde_json::json!(revision.version);
+                    observation["cluster_id"] = serde_json::json!(revision.cluster_id);
+                    observation["epoch"] = serde_json::json!(revision.epoch);
+                    observation["generation"] = serde_json::json!(revision.generation);
+                    observation["transaction_id"] = serde_json::json!(revision.transaction_id);
+                    observation["pool_count"] = serde_json::json!(meta.pools.len());
+                    observation["payload_sha256"] = serde_json::json!(rustfs_utils::crypto::hex(Sha256::digest(canonical)));
+                    observation["raw_sha256"] = serde_json::json!(rustfs_utils::crypto::hex(Sha256::digest(raw)));
+                }
+                PoolMetaReplica::Missing => observation["state"] = serde_json::json!("missing"),
+                PoolMetaReplica::Corrupt(_) => observation["state"] = serde_json::json!("corrupt"),
+                PoolMetaReplica::Incompatible(_) => observation["state"] = serde_json::json!("incompatible"),
+                PoolMetaReplica::Unreadable(_) => observation["state"] = serde_json::json!("unreadable"),
+            }
+            startup_cas_test_observe(observation);
+        }
+    }
+    reads
 }
 
 fn select_pool_meta_replicas_observing<R>(write_state: &mut PoolMetaWriteState, replicas: Vec<R>) -> Result<PoolMetaSelection>
@@ -5480,6 +5522,60 @@ fn pool_meta_cas_preconditions(token: &PoolMetaCasToken, object: &str) -> Result
     }
 }
 
+#[cfg(feature = "e2e-test-hooks")]
+struct StartupCasObservation {
+    attempt: uuid::Uuid,
+    phase: &'static str,
+    pools: Vec<usize>,
+}
+
+#[cfg(feature = "e2e-test-hooks")]
+tokio::task_local! {
+    static STARTUP_CAS_OBSERVATION: StartupCasObservation;
+}
+
+// This scope follows only the directly polled startup future. Spawned work
+// does not inherit it; receiver evidence retains its existing RPC tuple.
+#[cfg(feature = "e2e-test-hooks")]
+pub(crate) async fn startup_cas_test_scope<S, F: std::future::Future>(
+    attempt: uuid::Uuid,
+    phase: &'static str,
+    pools: &[Arc<S>],
+    future: F,
+) -> F::Output {
+    STARTUP_CAS_OBSERVATION
+        .scope(
+            StartupCasObservation {
+                attempt,
+                phase,
+                // These identities are never dereferenced or logged. The
+                // caller and operation keep the same pool Arcs alive.
+                pools: pools.iter().map(|pool| Arc::as_ptr(pool) as usize).collect(),
+            },
+            future,
+        )
+        .await
+}
+
+// Direct JSON diagnostics are independent of the startup tracing subscriber.
+#[cfg(feature = "e2e-test-hooks")]
+pub(crate) fn startup_cas_test_observe(mut observation: serde_json::Value) {
+    let Some(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_NONCE")
+        .ok()
+        .and_then(|value| uuid::Uuid::parse_str(&value).ok())
+    else {
+        return;
+    };
+    observation["nonce"] = serde_json::json!(nonce);
+    observation["pid"] = serde_json::json!(std::process::id());
+    let _ = STARTUP_CAS_OBSERVATION.try_with(|scope| {
+        observation["attempt"] = serde_json::json!(scope.attempt);
+        observation["startup_phase"] = serde_json::json!(scope.phase);
+    });
+    let line = format!("RUSTFS_E2E_STARTUP_CAS {observation}\n");
+    let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes());
+}
+
 async fn save_pool_meta_object_cas<S>(
     pool: Arc<S>,
     object: &str,
@@ -5500,13 +5596,43 @@ where
         ..Default::default()
     };
     fence.add_to_options(&mut opts);
+    #[cfg(feature = "e2e-test-hooks")]
+    let observation = std::env::var_os("RUSTFS_E2E_STARTUP_CAS_NONCE").map(|_| {
+        serde_json::json!({
+            "kind": "cas", "object": object, "phase": phase,
+            "pool": STARTUP_CAS_OBSERVATION.try_with(|scope| {
+                scope.pools.iter().position(|identity| *identity == Arc::as_ptr(&pool) as usize)
+            }).ok().flatten(),
+            "payload_sha256": rustfs_utils::crypto::hex(Sha256::digest(&data)),
+            "if_match": opts.http_preconditions.as_ref().and_then(|p| p.if_match.as_deref()),
+            "if_none_match": opts.http_preconditions.as_ref().and_then(|p| p.if_none_match.as_deref()),
+            "tail_drained": opts.write_completion == crate::object_api::WriteCompletion::TailDrained,
+            "no_lock": opts.no_lock,
+        })
+    });
     let result = save_config_with_opts_and_metadata(pool, object, data, &opts).await;
     if matches!(&result, Err(Error::PreconditionFailed)) {
         record_pool_meta_stale_write_rejection(phase);
     }
-    let object_info = result?;
-    fence.ensure_held()?;
-    Ok(object_info)
+    let result = result.and_then(|object_info| {
+        fence.ensure_held()?;
+        Ok(object_info)
+    });
+    #[cfg(feature = "e2e-test-hooks")]
+    if let Some(mut observation) = observation {
+        observation["ok"] = serde_json::json!(result.is_ok());
+        observation["etag"] = serde_json::json!(result.as_ref().ok().and_then(|info| info.etag.as_deref()));
+        observation["mod_time"] = serde_json::json!(
+            result
+                .as_ref()
+                .ok()
+                .and_then(|info| info.mod_time)
+                .map(|time| time.unix_timestamp_nanos().to_string())
+        );
+        observation["error"] = serde_json::json!(result.as_ref().err().map(ToString::to_string));
+        startup_cas_test_observe(observation);
+    }
+    result
 }
 
 async fn persist_pool_meta_identity<S>(
@@ -6805,6 +6931,13 @@ impl PoolMeta {
         };
         if confirmed.revision == revision && confirmed.canonical.as_ref() == Some(&durable) {
             persist_pool_meta_identity(pools, write_state, true, fence).await?;
+            #[cfg(feature = "e2e-test-hooks")]
+            startup_cas_test_observe(serde_json::json!({
+                "kind": "confirmed", "object": POOL_META_NAME,
+                "payload_sha256": rustfs_utils::crypto::hex(Sha256::digest(&durable)),
+                "generation": confirmed.revision.generation,
+                "transaction_id": confirmed.revision.transaction_id,
+            }));
             return Ok(confirmed.meta);
         }
         if !commit_succeeded {

crates/ecstore/src/disk/os.rs

@@ -247,7 +247,7 @@ pub(crate) mod fsync_dir_recorder {
 }
 
 /// Pause a real namespace mutation inside its physical executor.
-#[cfg(all(test, not(windows)))]
+#[cfg(all(any(test, feature = "test-util"), not(windows)))]
 pub(crate) mod prepared_publication_test_hooks {
     use super::*;
 
@@ -256,7 +256,9 @@ pub(crate) mod prepared_publication_test_hooks {
         PreparedRename,
         Rename,
         Remove,
+        #[cfg(test)]
         Rollback,
+        #[cfg(test)]
         DirFsync,
     }
 
@@ -272,6 +274,7 @@ pub(crate) mod prepared_publication_test_hooks {
         }
     }
 
+    #[cfg(test)]
     pub(crate) fn install(path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard {
         install_at(Stage::PreparedRename, path, hook)
     }
@@ -290,6 +293,51 @@ pub(crate) mod prepared_publication_test_hooks {
     }
 }
 
+/// Controlled application-test pause at an existing physical executor boundary.
+#[cfg(all(feature = "test-util", not(windows)))]
+pub struct LocalPublicationPause {
+    _hook: prepared_publication_test_hooks::Guard,
+    entered: oneshot::Receiver<()>,
+    _release: std::sync::mpsc::Sender<()>,
+}
+
+#[cfg(all(feature = "test-util", not(windows)))]
+#[derive(Clone, Copy)]
+pub enum LocalPublicationStage {
+    PreparedRename,
+    Rename,
+    Remove,
+}
+
+#[cfg(all(feature = "test-util", not(windows)))]
+impl LocalPublicationPause {
+    pub fn install(disk: &crate::disk::Disk, volume: &str, path: &str, stage: LocalPublicationStage) -> Result<Self> {
+        let path = disk
+            .get_object_path_for_io_if_local(volume, path)
+            .ok_or(DiskError::DiskNotFound)??;
+        let stage = match stage {
+            LocalPublicationStage::PreparedRename => prepared_publication_test_hooks::Stage::PreparedRename,
+            LocalPublicationStage::Rename => prepared_publication_test_hooks::Stage::Rename,
+            LocalPublicationStage::Remove => prepared_publication_test_hooks::Stage::Remove,
+        };
+        let (entered_tx, entered) = oneshot::channel();
+        let (release, release_rx) = std::sync::mpsc::channel::<()>();
+        let hook = prepared_publication_test_hooks::install_at(stage, &path, move || {
+            let _ = entered_tx.send(());
+            let _ = release_rx.recv();
+        });
+        Ok(Self {
+            _hook: hook,
+            entered,
+            _release: release,
+        })
+    }
+
+    pub async fn entered(&mut self) -> std::result::Result<(), oneshot::error::RecvError> {
+        (&mut self.entered).await
+    }
+}
+
 #[cfg(all(test, windows))]
 pub(crate) mod windows_rename_test_hooks {
     use super::*;
@@ -1916,7 +1964,7 @@ pub(crate) async fn remove_file_with_owner(
     let path = path.as_ref().to_path_buf();
     let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
     run_blocking_namespace_operation(lease, move || {
-        #[cfg(all(test, not(windows)))]
+        #[cfg(all(any(test, feature = "test-util"), not(windows)))]
         prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Remove, &path);
         std::fs::remove_file(path)
     })
@@ -2140,7 +2188,7 @@ pub(crate) async fn rename_all_with_prepared_source(
         move || {
             validate_prepared_rename_source(&prepared_source, &src_file_path)?;
             let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?;
-            #[cfg(test)]
+            #[cfg(any(test, feature = "test-util"))]
             prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path);
             rename_prepared(&src_file_path, &dst_file_path, &preparation)
         }
@@ -2271,7 +2319,7 @@ async fn reliable_rename_inner_with_lease(
         let base_dir = base_dir.clone();
         move || {
             let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?;
-            #[cfg(all(test, not(windows)))]
+            #[cfg(all(any(test, feature = "test-util"), not(windows)))]
             {
                 prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src_file_path);
                 prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path);

crates/ecstore/src/store/init.rs

@@ -630,14 +630,27 @@ impl ECStore {
             .pools
             .first()
             .is_some_and(|pool| pool_first_endpoint_is_local(&pool.endpoints));
+        #[cfg(feature = "e2e-test-hooks")]
+        let startup_attempt = uuid::Uuid::new_v4();
         let (meta, pool_meta_replica_state) = {
             let mut write_state = self.pool_meta_save_gate.lock().await;
             establish_pool_meta_bootstrap_identity_if_proven(self.pools.clone(), &mut write_state, should_persist_pool_meta)
                 .await
                 .map_err(|err| Error::other(format!("store init failed during establish_pool_meta_bootstrap_identity: {err}")))?;
-            load_pool_meta_for_startup(self.pools.clone(), &mut write_state).await?
+            let load = load_pool_meta_for_startup(self.pools.clone(), &mut write_state);
+            #[cfg(feature = "e2e-test-hooks")]
+            let load = crate::core::pools::startup_cas_test_scope(startup_attempt, "load", &self.pools, load);
+            load.await?
         };
         let update = meta.validate(self.pools.clone())?;
+        #[cfg(feature = "e2e-test-hooks")]
+        crate::core::pools::startup_cas_test_observe(serde_json::json!({
+            "kind": "startup-classifier", "attempt": startup_attempt,
+            "elected_writer": should_persist_pool_meta,
+            "needs_repair": pool_meta_replica_state.needs_repair,
+            "repair_write_safe": pool_meta_replica_state.repair_write_safe,
+            "topology_update": update,
+        }));
         let endpoints = runtime_sources::endpoint_pools_or_default();
 
         let mut installed_pool_meta = if update {
@@ -649,15 +662,17 @@ impl ECStore {
         // distributed startup can race on the same lock and replay the prior init bug.
         {
             let mut write_state = self.pool_meta_save_gate.lock().await;
-            installed_pool_meta = persist_pool_meta_for_startup_if_safe(
+            let persist = persist_pool_meta_for_startup_if_safe(
                 &installed_pool_meta,
                 self.pools.clone(),
                 pool_meta_replica_state,
                 &mut write_state,
                 update,
                 should_persist_pool_meta,
-            )
-            .await?;
+            );
+            #[cfg(feature = "e2e-test-hooks")]
+            let persist = crate::core::pools::startup_cas_test_scope(startup_attempt, "persist", &self.pools, persist);
+            installed_pool_meta = persist.await?;
         }
 
         {

crates/ecstore/src/store/mod.rs

@@ -442,7 +442,7 @@ pub(crate) mod utils;
 
 use peer::init_local_peer;
 pub use peer::{
-    all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks,
+    BootstrapLocalTarget, all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks,
     init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
     prewarm_local_disk_id_map_with_instance_ctx,
 };
@@ -1787,7 +1787,7 @@ mod tests {
 
     // Build a minimal ECStore carrying an explicit instance context. Empty
     // pools/disks are sufficient: the Phase 5 accessors read only `self.ctx`.
-    fn build_store_with_ctx(ctx: Arc<InstanceContext>) -> Arc<ECStore> {
+    pub(super) fn build_store_with_ctx(ctx: Arc<InstanceContext>) -> Arc<ECStore> {
         let endpoint_pools = EndpointServerPools::default();
         Arc::new(ECStore {
             id: uuid::Uuid::new_v4(),

crates/ecstore/src/store/peer.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use super::*;
-use crate::runtime::instance::InstanceContext;
+use crate::bucket::utils::has_bad_path_component;
+use crate::disk::error::{DiskError, Result as DiskResult};
+use crate::disk::{DeleteOptions, Disk, RenameDataGuards, RenameDataResp};
+use crate::runtime::instance::{InstanceContext, NamespaceCommitGuard};
 use crate::runtime::sources as runtime_sources;
 use tracing::{debug, error};
 
@@ -22,6 +25,203 @@ const LOG_SUBSYSTEM_DISK_STARTUP: &str = "disk_startup";
 const EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED: &str = "local_disk_id_prewarm_skipped";
 const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initialization_failed";
 
+/// An instance-bound capability for internal writes before ECStore/IAM startup.
+/// Its private context and volume checks cannot be replaced by a caller guard.
+#[derive(Clone)]
+pub struct BootstrapLocalTarget {
+    ctx: Arc<InstanceContext>,
+}
+
+impl BootstrapLocalTarget {
+    pub fn new(ctx: Arc<InstanceContext>) -> Self {
+        Self { ctx }
+    }
+
+    pub fn is_for_store(&self, store: &ECStore) -> bool {
+        Arc::ptr_eq(&self.ctx, &store.ctx)
+    }
+
+    pub async fn rename_local_data(
+        &self,
+        disk_ref: &str,
+        source: (&str, &str),
+        fi: &FileInfo,
+        destination: (&str, &str),
+        scanner_token: Option<Uuid>,
+    ) -> DiskResult<RenameDataResp> {
+        if scanner_token.is_some() {
+            return Err(DiskError::other("bootstrap rename cannot use a scanner publication lease"));
+        }
+        validate_bootstrap_volume(source.0)?;
+        validate_bootstrap_volume(destination.0)?;
+        rename_local_data_with_ctx(&self.ctx, disk_ref, source, fi, destination, RenameDataGuards::default()).await
+    }
+
+    pub async fn undo_local_write(
+        &self,
+        disk_ref: &str,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        opts: DeleteOptions,
+    ) -> DiskResult<()> {
+        validate_bootstrap_volume(volume)?;
+        undo_local_write_with_ctx(&self.ctx, disk_ref, volume, path, fi, opts).await
+    }
+}
+
+fn validate_bootstrap_volume(volume: &str) -> DiskResult<()> {
+    // Prefix membership alone permits aliases such as .rustfs.sys/../bucket.
+    // Validate both raw rename volumes before any disk lookup or admission.
+    if has_bad_path_component(volume) || !is_meta_bucketname(volume) {
+        return Err(DiskError::FileAccessDenied);
+    }
+    Ok(())
+}
+
+impl ECStore {
+    /// Execute on this instance's active local disk through the physical owner.
+    pub async fn rename_local_data(
+        &self,
+        disk_ref: &str,
+        source: (&str, &str),
+        fi: &FileInfo,
+        destination: (&str, &str),
+        scanner_token: Option<Uuid>,
+    ) -> DiskResult<RenameDataResp> {
+        let external_guard: Option<Arc<dyn Send + Sync>> = if let Some(token) = scanner_token {
+            Some(Arc::new(
+                self.acquire_scanner_publication_lease_guard(token)
+                    .await
+                    .map_err(|err| DiskError::other(err.to_string()))?,
+            ))
+        } else {
+            None
+        };
+        rename_local_data_with_ctx(
+            &self.ctx,
+            disk_ref,
+            source,
+            fi,
+            destination,
+            RenameDataGuards {
+                scanner_publication_lease_token: scanner_token,
+                external_guard,
+                namespace_owner: None,
+            },
+        )
+        .await
+    }
+
+    pub async fn undo_local_write(
+        &self,
+        disk_ref: &str,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        opts: DeleteOptions,
+    ) -> DiskResult<()> {
+        undo_local_write_with_ctx(&self.ctx, disk_ref, volume, path, fi, opts).await
+    }
+}
+
+// The optional ID is a cold lookup to cache only after final admission.
+async fn local_disk_candidate(ctx: &Arc<InstanceContext>, disk_ref: &str) -> DiskResult<(DiskStore, Option<Uuid>)> {
+    let map = ctx.local_disk_map();
+    if let Some(disk) = map.read().await.get(disk_ref).and_then(Option::as_ref).cloned() {
+        return Ok((disk, None));
+    }
+    let disk_id = Uuid::parse_str(disk_ref).map_err(|_| DiskError::DiskNotFound)?;
+    let cached_path = ctx.local_disk_id_map().read().await.get(&disk_id).cloned();
+    if let Some(path) = cached_path {
+        let cached_disk = map.read().await.get(&path).and_then(Option::as_ref).cloned();
+        if let Some(disk) = cached_disk
+            && matches!(disk.as_ref(), Disk::Local(_))
+            && disk.get_disk_id().await? == Some(disk_id)
+        {
+            return Ok((disk, None));
+        }
+    }
+    let disks: Vec<_> = map.read().await.values().filter_map(Clone::clone).collect();
+    // Disk identity may perform format I/O. No registry guard spans this await.
+    for disk in disks {
+        if matches!(disk.as_ref(), Disk::Local(_)) && disk.get_disk_id().await.ok().flatten() == Some(disk_id) {
+            return Ok((disk, Some(disk_id)));
+        }
+    }
+    Err(DiskError::DiskNotFound)
+}
+
+async fn admit_local_disk(
+    ctx: &Arc<InstanceContext>,
+    disk: &DiskStore,
+    disk_id: Option<Uuid>,
+    volume: &str,
+) -> DiskResult<Option<Arc<NamespaceCommitGuard>>> {
+    if !matches!(disk.as_ref(), Disk::Local(_)) {
+        return Err(DiskError::DiskNotFound);
+    }
+    let map = ctx.local_disk_map();
+    let active = map.read().await;
+    if !active
+        .get(&disk.endpoint().to_string())
+        .and_then(Option::as_ref)
+        .is_some_and(|current| Arc::ptr_eq(current, disk))
+    {
+        return Err(DiskError::DiskNotFound);
+    }
+    // Preserve registry -> ID-cache lock order; no filesystem I/O under either.
+    if let Some(disk_id) = disk_id {
+        ctx.local_disk_id_map()
+            .write()
+            .await
+            .insert(disk_id, disk.endpoint().to_string());
+    }
+    // Admission linearizes under the registry read: replacement/quarantine
+    // before this point rejects; later changes do not revoke physical I/O.
+    Ok((!is_meta_bucketname(volume)).then(|| ctx.begin_namespace_commit()))
+}
+
+async fn rename_local_data_with_ctx(
+    ctx: &Arc<InstanceContext>,
+    disk_ref: &str,
+    source: (&str, &str),
+    fi: &FileInfo,
+    destination: (&str, &str),
+    mut guards: RenameDataGuards,
+) -> DiskResult<RenameDataResp> {
+    let (disk, disk_id) = local_disk_candidate(ctx, disk_ref).await?;
+    let owner = admit_local_disk(ctx, &disk, disk_id, destination.0).await?;
+    guards.namespace_owner = owner.as_ref().map(|owner| owner.clone() as Arc<dyn Send + Sync>);
+    let result = disk
+        .rename_data_borrowed_with_fence_observed(source.0, source.1, fi, destination.0, destination.1, guards)
+        .await
+        .result;
+    drop(owner);
+    result
+}
+
+async fn undo_local_write_with_ctx(
+    ctx: &Arc<InstanceContext>,
+    disk_ref: &str,
+    volume: &str,
+    path: &str,
+    fi: FileInfo,
+    opts: DeleteOptions,
+) -> DiskResult<()> {
+    if !opts.undo_write {
+        return Err(DiskError::other("target undo requires undo_write"));
+    }
+    let (disk, disk_id) = local_disk_candidate(ctx, disk_ref).await?;
+    let owner = admit_local_disk(ctx, &disk, disk_id, volume).await?;
+    let physical_owner = owner.as_ref().map(|owner| owner.clone() as Arc<dyn Send + Sync>);
+    let result = disk
+        .undo_write_with_namespace_owner(volume, path, fi, opts, physical_owner)
+        .await;
+    drop(owner);
+    result
+}
+
 async fn remember_local_disk_id(disk: &DiskStore) -> Option<Uuid> {
     remember_local_disk_id_with_instance_ctx(&crate::runtime::global::current_ctx(), disk).await
 }
@@ -265,6 +465,522 @@ mod tests {
         }])
     }
 
+    async fn target_disk(ctx: &Arc<InstanceContext>, root: &std::path::Path, id: Uuid) -> DiskStore {
+        let mut format = crate::layout::format::FormatV3::new(1, 1);
+        format.erasure.this = id;
+        format.erasure.sets[0][0] = id;
+        let meta = root.join(crate::disk::RUSTFS_META_BUCKET);
+        tokio::fs::create_dir_all(&meta).await.expect("create format volume");
+        tokio::fs::write(
+            meta.join(crate::disk::FORMAT_CONFIG_FILE),
+            serde_json::to_vec(&format).expect("encode format"),
+        )
+        .await
+        .expect("write real disk identity");
+        let mut endpoint = Endpoint::try_from(root.to_str().expect("UTF-8 root")).expect("endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("open real local disk");
+        assert_eq!(disk.get_disk_id().await.expect("read disk format identity"), Some(id));
+        ctx.local_disk_map()
+            .write()
+            .await
+            .insert(disk.endpoint().to_string(), Some(disk.clone()));
+        disk
+    }
+
+    fn target_file_info(object: &str, version: Uuid, body: &'static [u8]) -> FileInfo {
+        let mut fi = FileInfo::new(object, 1, 0);
+        fi.erasure.index = 1;
+        fi.version_id = Some(version);
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fi.size = i64::try_from(body.len()).expect("fixture length");
+        fi.parts = vec![rustfs_filemeta::ObjectPartInfo {
+            number: 1,
+            size: body.len(),
+            actual_size: fi.size,
+            ..Default::default()
+        }];
+        fi.data = Some(bytes::Bytes::from_static(body));
+        fi.set_inline_data();
+        fi
+    }
+
+    async fn seed_target(disk: &DiskStore, volume: &str, object: &str, fi: FileInfo) -> Vec<u8> {
+        let dir = disk.path().join(volume);
+        tokio::fs::create_dir_all(&dir).await.expect("real fixture volume");
+        disk.write_metadata(volume, volume, object, fi.clone())
+            .await
+            .expect("seed real metadata");
+        let read = disk
+            .read_version(
+                volume,
+                volume,
+                object,
+                &fi.version_id.expect("fixture version").to_string(),
+                &crate::disk::ReadOptions {
+                    read_data: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("read fixture before mutation");
+        assert_eq!(read.data, fi.data, "fixture must contain readable inline bytes");
+        tokio::fs::read(dir.join(object).join(crate::disk::STORAGE_FORMAT_FILE))
+            .await
+            .expect("seeded metadata bytes")
+    }
+
+    #[tokio::test]
+    async fn target_uuid_lookup_binds_real_disk_and_owner_to_one_instance() {
+        for warm in [false, true] {
+            let ctx_a = Arc::new(InstanceContext::new());
+            let ctx_b = Arc::new(InstanceContext::new());
+            let a = tempfile::tempdir().expect("A root");
+            let b = tempfile::tempdir().expect("B root");
+            let id = Uuid::new_v4();
+            let disk_a = target_disk(&ctx_a, a.path(), id).await;
+            let disk_b = target_disk(&ctx_b, b.path(), id).await;
+            if warm {
+                assert!(record_local_disk_id_if_active(&ctx_a, &disk_a, id).await);
+                assert!(record_local_disk_id_if_active(&ctx_b, &disk_b, id).await);
+            }
+            let version = Uuid::new_v4();
+            let fi = target_file_info("destination", version, b"new-A");
+            for disk in [&disk_a, &disk_b] {
+                seed_target(disk, "target-bucket", "staged", fi.clone()).await;
+            }
+            let b_before = seed_target(
+                &disk_b,
+                "target-bucket",
+                "destination",
+                target_file_info("destination", version, b"old-B"),
+            )
+            .await;
+            let store = super::super::tests::build_store_with_ctx(ctx_a.clone());
+            store
+                .rename_local_data(&id.to_string(), ("target-bucket", "staged"), &fi, ("target-bucket", "destination"), None)
+                .await
+                .expect("rename on A");
+            let read = disk_a
+                .read_version(
+                    "target-bucket",
+                    "target-bucket",
+                    "destination",
+                    &version.to_string(),
+                    &crate::disk::ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("read committed A");
+            assert_eq!(read.data, fi.data, "warm={warm}");
+            assert_eq!(
+                tokio::fs::read(b.path().join("target-bucket/destination/xl.meta"))
+                    .await
+                    .expect("B metadata"),
+                b_before
+            );
+            assert!(b.path().join("target-bucket/staged/xl.meta").exists());
+            assert!(ctx_a.namespace_commit_generation() > 0);
+            assert_eq!(ctx_b.namespace_commit_generation(), 0);
+            assert!(!ctx_a.namespace_commits_pending());
+            assert!(!ctx_b.namespace_commits_pending());
+            assert_eq!(ctx_a.local_disk_id_map().read().await.get(&id), Some(&disk_a.endpoint().to_string()));
+        }
+    }
+
+    #[tokio::test]
+    async fn target_admission_rejects_removed_quarantined_and_replaced_arcs() {
+        let ctx = Arc::new(InstanceContext::new());
+        let root = tempfile::tempdir().expect("root");
+        let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+        let endpoint = disk.endpoint().to_string();
+        for state in ["removed", "quarantined", "replaced"] {
+            let replacement = new_disk(
+                &disk.endpoint(),
+                &DiskOption {
+                    cleanup: false,
+                    health_check: false,
+                },
+            )
+            .await
+            .expect("separate active Arc");
+            let map = ctx.local_disk_map();
+            let mut entries = map.write().await;
+            match state {
+                "removed" => {
+                    entries.remove(&endpoint);
+                }
+                "quarantined" => {
+                    entries.insert(endpoint.clone(), None);
+                }
+                _ => {
+                    entries.insert(endpoint.clone(), Some(replacement));
+                }
+            }
+            drop(entries);
+            assert!(
+                matches!(admit_local_disk(&ctx, &disk, None, "target-bucket").await, Err(DiskError::DiskNotFound)),
+                "{state}"
+            );
+            assert!(!ctx.namespace_commits_pending());
+            assert_eq!(ctx.namespace_commit_generation(), 0);
+        }
+    }
+
+    #[tokio::test]
+    async fn target_uuid_cache_cannot_admit_a_different_format_at_the_same_path() {
+        let ctx = Arc::new(InstanceContext::new());
+        let root = tempfile::tempdir().expect("root");
+        let old_id = Uuid::new_v4();
+        let old = target_disk(&ctx, root.path(), old_id).await;
+        assert!(record_local_disk_id_if_active(&ctx, &old, old_id).await);
+        let replacement_id = Uuid::new_v4();
+        let replacement = target_disk(&ctx, root.path(), replacement_id).await;
+        assert!(!Arc::ptr_eq(&old, &replacement));
+        assert!(matches!(
+            local_disk_candidate(&ctx, &old_id.to_string()).await,
+            Err(DiskError::DiskNotFound)
+        ));
+        let (candidate, verified) = local_disk_candidate(&ctx, &replacement_id.to_string())
+            .await
+            .expect("replacement UUID");
+        assert!(Arc::ptr_eq(&candidate, &replacement));
+        assert_eq!(verified, Some(replacement_id));
+        assert!(!ctx.namespace_commits_pending());
+    }
+
+    #[tokio::test]
+    async fn bootstrap_rejects_user_volumes_aliases_and_scanner_tokens_without_mutation() {
+        let ctx = Arc::new(InstanceContext::new());
+        let root = tempfile::tempdir().expect("root");
+        let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+        let target = BootstrapLocalTarget::new(ctx.clone());
+        let fi = target_file_info("destination", Uuid::new_v4(), b"body");
+        let user_before = seed_target(&disk, "victim", "staged", fi.clone()).await;
+        let meta_before = seed_target(&disk, ".rustfs.sys/tmp", "staged", fi.clone()).await;
+        for invalid in [
+            "victim",
+            ".rustfs.sys/../victim",
+            ".rustfs.sys/./tmp",
+            ".rustfs.sys/ .. /victim",
+            ".rustfs.sys\\..\\victim",
+            ".minio.sys/../victim",
+        ] {
+            for (src, dst) in [(invalid, ".rustfs.sys/tmp"), (".rustfs.sys/tmp", invalid)] {
+                assert!(
+                    target
+                        .rename_local_data(&disk.endpoint().to_string(), (src, "staged"), &fi, (dst, "destination"), None)
+                        .await
+                        .is_err(),
+                    "src={src}, dst={dst}"
+                );
+            }
+            assert!(
+                target
+                    .undo_local_write(
+                        &disk.endpoint().to_string(),
+                        invalid,
+                        "staged",
+                        fi.clone(),
+                        DeleteOptions {
+                            undo_write: true,
+                            ..Default::default()
+                        }
+                    )
+                    .await
+                    .is_err(),
+                "{invalid}"
+            );
+        }
+        assert!(
+            target
+                .rename_local_data(
+                    &disk.endpoint().to_string(),
+                    (".rustfs.sys/tmp", "staged"),
+                    &fi,
+                    (".rustfs.sys/tmp", "destination"),
+                    Some(Uuid::new_v4())
+                )
+                .await
+                .is_err()
+        );
+        assert_eq!(
+            tokio::fs::read(root.path().join("victim/staged/xl.meta"))
+                .await
+                .expect("user source"),
+            user_before
+        );
+        assert_eq!(
+            tokio::fs::read(root.path().join(".rustfs.sys/tmp/staged/xl.meta"))
+                .await
+                .expect("metadata source"),
+            meta_before
+        );
+        assert!(!root.path().join("victim/destination").exists());
+        assert!(!root.path().join(".rustfs.sys/tmp/destination").exists());
+        assert_eq!(ctx.namespace_commit_generation(), 0);
+        assert!(!ctx.namespace_commits_pending());
+    }
+
+    #[tokio::test]
+    async fn bootstrap_allows_internal_multisegment_rename_without_namespace_owner() {
+        for volume in [".rustfs.sys/tmp", ".rustfs.sys/multipart", ".minio.sys/config"] {
+            let ctx = Arc::new(InstanceContext::new());
+            let root = tempfile::tempdir().expect("root");
+            let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+            let fi = target_file_info("destination", Uuid::new_v4(), b"internal-CAS-body");
+            seed_target(&disk, volume, "staged", fi.clone()).await;
+            BootstrapLocalTarget::new(ctx.clone())
+                .rename_local_data(&disk.endpoint().to_string(), (volume, "staged"), &fi, (volume, "destination"), None)
+                .await
+                .expect("legitimate bootstrap metadata write");
+            let read = disk
+                .read_version(
+                    volume,
+                    volume,
+                    "destination",
+                    &fi.version_id.expect("version").to_string(),
+                    &crate::disk::ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("read bootstrap result");
+            assert_eq!(read.data, fi.data);
+            assert_eq!(ctx.namespace_commit_generation(), 0);
+            assert!(!ctx.namespace_commits_pending());
+        }
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn target_rename_cancellation_retains_real_namespace_and_scanner_owners() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+        let ctx = Arc::new(InstanceContext::new());
+        let sibling = Arc::new(InstanceContext::new());
+        let root = tempfile::tempdir().expect("root");
+        let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+        let store = super::super::tests::build_store_with_ctx(ctx.clone());
+        let fi = target_file_info("destination", Uuid::new_v4(), b"physically-owned");
+        seed_target(&disk, "target-bucket", "staged", fi.clone()).await;
+        let (token, _) = store
+            .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL)
+            .await
+            .expect("real scanner token in A");
+        let destination = disk
+            .get_object_path_for_io_if_local("target-bucket", "destination/xl.meta")
+            .expect("local disk")
+            .expect("destination IO path");
+        let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        let _hook = hooks::install(&destination, move || {
+            let _ = entered_tx.send(());
+            let _ = release_rx.recv();
+        });
+        let disk_ref = disk.endpoint().to_string();
+        let mut rename = Box::pin(store.rename_local_data(
+            &disk_ref,
+            ("target-bucket", "staged"),
+            &fi,
+            ("target-bucket", "destination"),
+            Some(token),
+        ));
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            tokio::select! {
+                result = &mut rename => panic!("rename completed before physical pause: {result:?}"),
+                entered = entered_rx => entered.expect("physical rename entered"),
+            }
+        })
+        .await
+        .expect("bounded physical entry");
+        drop(rename);
+        assert!(store.scanner_data_usage_publication_blocked().await);
+        assert!(ctx.namespace_commits_pending());
+        assert!(!sibling.namespace_commits_pending());
+        assert!(
+            store
+                .rename_local_data(&disk_ref, ("target-bucket", "staged"), &fi, ("target-bucket", "another"), Some(token))
+                .await
+                .is_err(),
+            "real pending rename blocks another scanner publication"
+        );
+        assert!(store.release_scanner_publication_lease(token).await, "remove registered token");
+        let gate = ctx.data_movement_operation_gate();
+        assert!(
+            gate.clone().try_write_owned().is_err(),
+            "physical operation still owns the scanner read guard"
+        );
+        drop(release_tx);
+        let _drained = tokio::time::timeout(std::time::Duration::from_secs(10), gate.write_owned())
+            .await
+            .expect("physical tail must release scanner guard");
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            while ctx.namespace_commits_pending() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("namespace owner drains");
+        let read = disk
+            .read_version(
+                "target-bucket",
+                "target-bucket",
+                "destination",
+                &fi.version_id.expect("version").to_string(),
+                &crate::disk::ReadOptions {
+                    read_data: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("read actual late commit");
+        assert_eq!(read.data, fi.data);
+        assert!(ctx.namespace_commit_generation() >= 2);
+        assert_eq!(sibling.namespace_commit_generation(), 0);
+    }
+
+    #[tokio::test]
+    async fn target_ready_rejects_unknown_foreign_released_and_expired_scanner_tokens() {
+        let ctx = Arc::new(InstanceContext::new());
+        let other = Arc::new(InstanceContext::new());
+        let store = super::super::tests::build_store_with_ctx(ctx.clone());
+        let other_store = super::super::tests::build_store_with_ctx(other);
+        let root = tempfile::tempdir().expect("root");
+        let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+        let fi = target_file_info("destination", Uuid::new_v4(), b"unchanged");
+        let before = seed_target(&disk, "target-bucket", "staged", fi.clone()).await;
+        let ttl = crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL;
+        let (foreign, _) = other_store.acquire_scanner_publication_lease(0, ttl).await.expect("B token");
+        let (released, _) = store.acquire_scanner_publication_lease(0, ttl).await.expect("A token");
+        assert!(store.release_scanner_publication_lease(released).await);
+        let (valid, _) = store.acquire_scanner_publication_lease(0, ttl).await.expect("new A token");
+        for token in [Uuid::new_v4(), foreign, released] {
+            assert!(
+                store
+                    .rename_local_data(
+                        &disk.endpoint().to_string(),
+                        ("target-bucket", "staged"),
+                        &fi,
+                        ("target-bucket", "destination"),
+                        Some(token)
+                    )
+                    .await
+                    .is_err()
+            );
+        }
+        tokio::time::pause();
+        tokio::time::advance(ttl + std::time::Duration::from_secs(1)).await;
+        tokio::time::resume();
+        assert!(
+            store
+                .rename_local_data(
+                    &disk.endpoint().to_string(),
+                    ("target-bucket", "staged"),
+                    &fi,
+                    ("target-bucket", "destination"),
+                    Some(valid)
+                )
+                .await
+                .is_err(),
+            "expired real token"
+        );
+        let _ = other_store.release_scanner_publication_lease(foreign).await;
+        assert_eq!(
+            tokio::fs::read(root.path().join("target-bucket/staged/xl.meta"))
+                .await
+                .expect("source bytes"),
+            before
+        );
+        assert!(!root.path().join("target-bucket/destination").exists());
+        assert!(!ctx.namespace_commits_pending());
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn target_ordinary_timeout_keeps_its_physical_namespace_owner() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("1"))], async {
+            let ctx = Arc::new(InstanceContext::new());
+            let store = super::super::tests::build_store_with_ctx(ctx.clone());
+            let root = tempfile::tempdir().expect("root");
+            let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+            let fi = target_file_info("destination", Uuid::new_v4(), b"timed-out-physical-commit");
+            seed_target(&disk, "target-bucket", "staged", fi.clone()).await;
+            let path = disk
+                .get_object_path_for_io_if_local("target-bucket", "destination/xl.meta")
+                .expect("local")
+                .expect("destination IO path");
+            let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+            let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+            let _hook = hooks::install(&path, move || {
+                let _ = entered_tx.send(());
+                let _ = release_rx.recv();
+            });
+            let disk_ref = disk.endpoint().to_string();
+            let mut rename = Box::pin(store.rename_local_data(
+                &disk_ref,
+                ("target-bucket", "staged"),
+                &fi,
+                ("target-bucket", "destination"),
+                None,
+            ));
+            tokio::time::timeout(std::time::Duration::from_secs(10), async {
+                tokio::select! {
+                    result = &mut rename => panic!("completed before physical pause: {result:?}"),
+                    entered = entered_rx => entered.expect("physical entry"),
+                }
+            })
+            .await
+            .expect("bounded entry");
+            tokio::time::pause();
+            tokio::time::advance(std::time::Duration::from_secs(2)).await;
+            tokio::time::resume();
+            let result = tokio::time::timeout(std::time::Duration::from_secs(5), &mut rename)
+                .await
+                .expect("ordinary deadline remains enabled");
+            assert!(matches!(result, Err(DiskError::Timeout)), "{result:?}");
+            drop(rename);
+            assert!(ctx.namespace_commits_pending(), "timeout is not a physical drain");
+            drop(release_tx);
+            tokio::time::timeout(std::time::Duration::from_secs(10), async {
+                while ctx.namespace_commits_pending() {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("late physical owner drains");
+            let read = disk
+                .read_version(
+                    "target-bucket",
+                    "target-bucket",
+                    "destination",
+                    &fi.version_id.expect("version").to_string(),
+                    &crate::disk::ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("read actual timeout tail");
+            assert_eq!(read.data, fi.data);
+        })
+        .await;
+    }
+
     #[test]
     fn endpoint_rpc_authority_preserves_port_and_ipv6_brackets() {
         let endpoint = Endpoint::try_from("https://127.0.0.1:9001/d1").expect("URL endpoint");

rustfs/Cargo.toml

@@ -68,7 +68,7 @@ license = []
 io-scheduler-debug = []  # Enable debug information in I/O scheduler
 tracing-chunk-debug = []  # Enable per-chunk tracing in data plane (high noise, for debugging only)
 full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp", "pyroscope", "gcs"]
-e2e-test-hooks = []
+e2e-test-hooks = ["rustfs-ecstore/e2e-test-hooks"]
 # Shortens Connect credentials only in debug E2E builds.
 connect-e2e-short-credentials = []
 # Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root.

rustfs/src/app/context/server_slot.rs

@@ -26,13 +26,14 @@
 //! server is not ready rather than that another server's global context applies.
 
 use super::global::{AppContext, get_global_app_context};
-use crate::app::storage_api::context::ECStore;
+use crate::app::storage_api::context::{BootstrapLocalTarget, ECStore, InstanceContext};
 use std::sync::{Arc, OnceLock};
 
 /// Late-bound, per-server handle to the application context.
 #[derive(Default)]
 pub struct ServerContextSlot {
     app_context: OnceLock<Arc<AppContext>>,
+    bootstrap_target: Option<BootstrapLocalTarget>,
     heal_topology_fingerprint: Arc<tokio::sync::OnceCell<String>>,
 }
 
@@ -50,15 +51,47 @@ impl ServerContextSlot {
     pub fn new() -> Arc<Self> {
         Arc::new(Self {
             app_context: OnceLock::new(),
+            bootstrap_target: None,
             heal_topology_fingerprint: Arc::new(tokio::sync::OnceCell::new()),
         })
     }
 
+    /// Bind the listener to its foundation before it can accept requests.
+    pub fn with_instance_context(ctx: Arc<InstanceContext>) -> Arc<Self> {
+        Arc::new(Self {
+            bootstrap_target: Some(BootstrapLocalTarget::new(ctx)),
+            ..Self::default()
+        })
+    }
+
     /// Install this server's application context (once). Returns `false` if
     /// the slot was already installed; the first installation wins, matching
     /// the process-global singleton's `get_or_init` semantics.
     pub fn install(&self, context: Arc<AppContext>) -> bool {
-        self.app_context.set(context).is_ok()
+        self.try_install(context).is_ok()
+    }
+
+    /// Claim the slot before any process-global application publication.
+    /// Repeated installation, even of the same Arc, is an explicit conflict.
+    pub fn try_install(&self, context: Arc<AppContext>) -> std::io::Result<()> {
+        if self
+            .bootstrap_target
+            .as_ref()
+            .is_some_and(|target| !target.is_for_store(&context.object_store()))
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "application context does not belong to this server foundation",
+            ));
+        }
+        self.app_context.set(context).map_err(|_| {
+            std::io::Error::new(std::io::ErrorKind::AlreadyExists, "server application context is already installed")
+        })
+    }
+
+    /// Immutable, restricted startup capability; never resolves an ambient store.
+    pub fn bootstrap_target(&self) -> Option<BootstrapLocalTarget> {
+        self.bootstrap_target.clone()
     }
 
     /// This server's installed application context, if startup has completed.

rustfs/src/app/context/startup.rs

@@ -37,8 +37,8 @@ impl AppContext {
         // also publishes to the process default (first server wins) so legacy
         // free-function readers keep resolving the first server's context.
         let context = Arc::new(AppContext::with_default_interfaces(store, iam, kms_interface));
-        publish_global_app_context(context.clone());
-        let _ = server_ctx.install(context);
+        server_ctx.try_install(context.clone())?;
+        publish_global_app_context(context);
         Ok(())
     }
 }

rustfs/src/app/storage_api.rs

@@ -1261,7 +1261,7 @@ pub(crate) mod context {
     pub(crate) use super::EndpointServerPools;
     pub(crate) use super::bucket;
     pub(crate) use super::runtime;
-    pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools};
+    pub(crate) use crate::storage::storage_api::{BootstrapLocalTarget, ECStore, EndpointServerPools, InstanceContext};
     #[cfg(test)]
     pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints};
 }

rustfs/src/server/http.rs

@@ -36,7 +36,9 @@ use crate::server::{
 };
 use crate::storage_api::server::http as storage;
 use crate::storage_api::server::http::rpc::InternodeRpcService;
+#[cfg(test)]
 use crate::storage_api::server::http::tonic_service::make_server;
+use crate::storage_api::server::http::tonic_service::make_server_for_slot;
 use crate::storage_api::server::http::{
     ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge,
     tonic_boot_epoch_response_headers, verify_tonic_rpc_signature_with_bootstrap,
@@ -1834,7 +1836,7 @@ fn process_connection(
         // each service in the auth interceptor.
         let rpc_max_message_size = rustfs_protos::internode_rpc_max_message_size();
         let node_service = InterceptedService::new(
-            NodeServiceServer::new(make_server())
+            NodeServiceServer::new(make_server_for_slot(Arc::clone(&server_ctx)))
                 .max_decoding_message_size(rpc_max_message_size)
                 .max_encoding_message_size(rpc_max_message_size),
             check_auth,

rustfs/src/startup_embedded.rs

@@ -124,9 +124,6 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
     } else {
         bootstrap_instance_ctx()
     };
-    // This server's request-path context slot (backlog#1052 S2).
-    let server_ctx = ServerContextSlot::new();
-
     let EmbeddedStartupConfig {
         config,
         identity,
@@ -151,6 +148,7 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
         .await
         .map_err(init_error)?;
 
+    let server_ctx = ServerContextSlot::with_instance_context(instance_ctx.clone());
     let http_server = start_embedded_http_server(&config, listen_context.readiness.clone(), server_ctx.clone()).await?;
     let shutdown_handle = http_server.shutdown_handle;
     let bound_addr = http_server.bound_addr;

rustfs/src/startup_entrypoint.rs

@@ -62,6 +62,29 @@ fn emit_fatal_stderr(context: &str, error: impl std::fmt::Display) {
 }
 
 async fn async_main() -> Result<()> {
+    #[cfg(feature = "e2e-test-hooks")]
+    if let Ok(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_PROBE") {
+        let nonce = uuid::Uuid::parse_str(&nonce).map_err(Error::other)?;
+        // This precedes CLI parsing and observability, including `--help`.
+        println!(
+            "RUSTFS_E2E_STARTUP_CAS {}",
+            serde_json::json!({
+                "kind": "capability", "schema": "fresh-startup-cas/v1", "nonce": nonce,
+            })
+        );
+        return Ok(());
+    }
+    #[cfg(feature = "e2e-test-hooks")]
+    if let Ok(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_NONCE") {
+        let nonce = uuid::Uuid::parse_str(&nonce).map_err(Error::other)?;
+        let line = format!(
+            "RUSTFS_E2E_STARTUP_CAS {}\n",
+            serde_json::json!({
+                "kind": "observer-ready", "nonce": nonce, "pid": std::process::id(),
+            })
+        );
+        let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes());
+    }
     hotpath::tokio_runtime!();
 
     // Log container resource detection early in startup
@@ -141,10 +164,6 @@ async fn run(config: Config) -> Result<()> {
     // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future
     // multi-instance server constructs its own context here instead.
     let instance_ctx = bootstrap_instance_ctx();
-    // This server's request-path context slot (backlog#1052 S2): handed to the
-    // HTTP service now, installed once IAM bootstrap completes.
-    let server_ctx = ServerContextSlot::new();
-
     let StartupListenContext {
         readiness,
         server_addr,
@@ -152,6 +171,7 @@ async fn run(config: Config) -> Result<()> {
     } = init_startup_listen_context(&config, &instance_ctx).await?;
 
     let endpoint_pools = init_startup_storage_foundation(&server_address, &config.volumes, &instance_ctx).await?;
+    let server_ctx = ServerContextSlot::with_instance_context(instance_ctx.clone());
     let StartupHttpServers {
         state_manager,
         s3_shutdown_tx,
@@ -163,6 +183,33 @@ async fn run(config: Config) -> Result<()> {
         shutdown_token: ctx,
     } = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone(), instance_ctx).await?;
 
+    #[cfg(feature = "e2e-test-hooks")]
+    if let Ok(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_NONCE") {
+        let nonce = uuid::Uuid::parse_str(&nonce).map_err(Error::other)?;
+        let release = std::path::PathBuf::from(
+            std::env::var_os("RUSTFS_E2E_STARTUP_CAS_RELEASE")
+                .ok_or_else(|| Error::other("startup CAS fixture requires a release path"))?,
+        );
+        if server_ctx.installed_object_store().is_some() {
+            return Err(Error::other("startup CAS gate reached an installed slot"));
+        }
+        let line = format!(
+            "RUSTFS_E2E_STARTUP_CAS {}\n",
+            serde_json::json!({
+                "kind": "gate", "nonce": nonce, "pid": std::process::id(), "slot_installed": false,
+            })
+        );
+        let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes());
+        tokio::time::timeout(std::time::Duration::from_secs(180), async {
+            while !tokio::fs::try_exists(&release).await? {
+                tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+            }
+            Ok::<_, Error>(())
+        })
+        .await
+        .map_err(|_| Error::other("startup CAS gate release timed out"))??;
+    }
+
     let capacity_tasks = crate::capacity::capacity_integration::init_capacity_management_managed().await;
 
     let service_runtime = init_startup_runtime_services(

rustfs/src/storage/rpc/node_service.rs

@@ -28,9 +28,9 @@ use crate::storage::storage_api::rpc_consumer::node_service::{
     SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _,
     StorageResult, all_local_disk_path, find_local_disk_by_ref, reload_transition_tier_config,
 };
-use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, runtime_sources};
+use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, ServerContextSlot, runtime_sources};
 use crate::storage::storage_api::{
-    sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
+    BootstrapLocalTarget, sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
     verify_tonic_mutation_body_digest_reject_unsigned,
 };
 use bytes::Bytes;
@@ -482,6 +482,97 @@ mod metrics;
 pub struct NodeService {
     local_peer: LocalPeerS3Client,
     context: Option<Arc<runtime_sources::AppContext>>,
+    server_ctx: Option<Arc<ServerContextSlot>>,
+}
+
+enum LocalMutationTarget {
+    Ready(Arc<ECStore>),
+    Bootstrap(BootstrapLocalTarget),
+    Unbound,
+}
+
+#[cfg(feature = "e2e-test-hooks")]
+pub(crate) mod rename_target_capture_test_hook {
+    use super::LocalMutationTarget;
+    use rustfs_protos::proto_gen::node_service::RenameDataRequest;
+    use std::sync::{LazyLock, Mutex};
+    use tokio::sync::oneshot;
+    use uuid::Uuid;
+
+    struct Hook {
+        id: Uuid,
+        disk: String,
+        volume: String,
+        path: String,
+        captured: oneshot::Sender<bool>,
+        release: oneshot::Receiver<()>,
+    }
+
+    static HOOK: LazyLock<Mutex<Option<Hook>>> = LazyLock::new(|| Mutex::new(None));
+
+    /// One exact signed rename paused after its listener target was captured.
+    /// Dropping the handle removes an unused hook and releases an entered one.
+    pub struct RenameTargetCapturePause {
+        id: Uuid,
+        captured: oneshot::Receiver<bool>,
+        release: Option<oneshot::Sender<()>>,
+    }
+
+    impl RenameTargetCapturePause {
+        pub async fn wait_until_captured(&mut self) -> bool {
+            (&mut self.captured)
+                .await
+                .expect("matching rename must report its actual captured target")
+        }
+    }
+
+    impl Drop for RenameTargetCapturePause {
+        fn drop(&mut self) {
+            let unused = HOOK
+                .lock()
+                .expect("rename capture hook lock")
+                .take_if(|hook| hook.id == self.id);
+            drop(unused);
+            if let Some(release) = self.release.take() {
+                let _ = release.send(());
+            }
+        }
+    }
+
+    pub fn pause_rename_after_target_capture(disk: &str, volume: &str, path: &str) -> RenameTargetCapturePause {
+        let id = Uuid::new_v4();
+        let (captured_tx, captured) = oneshot::channel();
+        let (release, release_rx) = oneshot::channel();
+        let mut active = HOOK.lock().expect("rename capture hook lock");
+        if active.is_some() {
+            drop(active);
+            panic!("only one rename capture hook may be active");
+        }
+        *active = Some(Hook {
+            id,
+            disk: disk.to_owned(),
+            volume: volume.to_owned(),
+            path: path.to_owned(),
+            captured: captured_tx,
+            release: release_rx,
+        });
+        RenameTargetCapturePause {
+            id,
+            captured,
+            release: Some(release),
+        }
+    }
+
+    pub(super) async fn wait(target: &LocalMutationTarget, request: &RenameDataRequest) {
+        let hook = {
+            let mut active = HOOK.lock().expect("rename capture hook lock");
+            active.take_if(|hook| hook.disk == request.disk && hook.volume == request.dst_volume && hook.path == request.dst_path)
+        };
+        if let Some(hook) = hook {
+            let _ = hook.captured.send(matches!(target, LocalMutationTarget::Bootstrap(_)));
+            let _ = hook.release.await;
+        }
+    }
 }
 
 impl std::fmt::Debug for NodeService {
@@ -507,7 +598,19 @@ pub fn make_server() -> NodeService {
 
 pub fn make_server_for_context(context: Option<Arc<runtime_sources::AppContext>>) -> NodeService {
     let local_peer = LocalPeerS3Client::new(None, None);
-    NodeService { local_peer, context }
+    NodeService {
+        local_peer,
+        context,
+        server_ctx: None,
+    }
+}
+
+pub(crate) fn make_server_for_slot(server_ctx: Arc<ServerContextSlot>) -> NodeService {
+    // Unrelated RPCs retain their existing context policy. Target mutations
+    // resolve exclusively through this listener slot on each request.
+    let mut service = make_server();
+    service.server_ctx = Some(server_ctx);
+    service
 }
 
 #[derive(Clone, Debug, Default)]
@@ -1074,6 +1177,24 @@ impl heal_control_service_server::HealControlService for HealControlRpcService {
 }
 
 impl NodeService {
+    fn local_mutation_target(&self) -> LocalMutationTarget {
+        if let Some(slot) = &self.server_ctx {
+            // Capture exactly once per request, not at connection acceptance.
+            // A captured Bootstrap request cannot upgrade across a later await.
+            if let Some(store) = slot.installed_object_store() {
+                LocalMutationTarget::Ready(store)
+            } else if let Some(target) = slot.bootstrap_target() {
+                LocalMutationTarget::Bootstrap(target)
+            } else {
+                LocalMutationTarget::Unbound
+            }
+        } else if let Some(context) = &self.context {
+            LocalMutationTarget::Ready(context.object_store())
+        } else {
+            LocalMutationTarget::Unbound
+        }
+    }
+
     fn resolve_object_store(&self) -> Option<Arc<ECStore>> {
         let context = self.context.clone().or_else(runtime_sources::current_app_context);
         runtime_sources::current_object_store_handle_for_context(context.as_deref())
@@ -2680,6 +2801,7 @@ mod tests {
         validate_admin_heal_control_start,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
+    use crate::storage::storage_api::ecstore_disk::DiskAPI as _;
     use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo};
     use crate::storage::storage_api::set_tonic_canonical_body_digest;
     use crate::storage::storage_api::{
@@ -4660,6 +4782,687 @@ mod tests {
         assert!(rename_response.error.is_some());
     }
 
+    struct TargetRpcFixture {
+        _root: tempfile::TempDir,
+        env: rustfs_test_utils::TestECStoreEnv,
+        instance: Arc<crate::storage::storage_api::InstanceContext>,
+        context: Arc<crate::runtime_sources::AppContext>,
+        iam: Arc<rustfs_iam::sys::IamSys<ObjectStore>>,
+    }
+
+    async fn target_rpc_fixture() -> TargetRpcFixture {
+        super::timeout(Duration::from_secs(90), async {
+            let root = tempfile::tempdir().expect("target RPC root");
+            let env = rustfs_test_utils::TestECStoreEnv::builder()
+                .base_dir(root.path())
+                .init_bucket_metadata(false)
+                .build()
+                .await;
+            ObjectStore::new(env.ecstore.clone())
+                .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
+                .await
+                .expect("seed real IAM format");
+            let iam = rustfs_iam::build_iam_sys(env.ecstore.clone())
+                .await
+                .expect("build fixture IAM");
+            let context = Arc::new(crate::runtime_sources::AppContext::with_default_interfaces(
+                env.ecstore.clone(),
+                iam.clone(),
+                Arc::new(KmsServiceManager::new()),
+            ));
+            let instance = crate::storage::storage_api::bootstrap_instance_ctx();
+            assert!(
+                super::BootstrapLocalTarget::new(instance.clone()).is_for_store(&env.ecstore),
+                "the standard builder must use this exact instance context"
+            );
+            super::timeout(Duration::from_secs(10), async {
+                while env.ecstore.scanner_data_usage_publication_blocked().await {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("startup namespace commits drain before test");
+            TargetRpcFixture {
+                _root: root,
+                env,
+                instance,
+                context,
+                iam,
+            }
+        })
+        .await
+        .expect("bounded real fixture initialization")
+    }
+
+    async fn stage_target_rpc(fixture: &TargetRpcFixture) -> (super::DiskStore, rustfs_filemeta::FileInfo, Vec<u8>) {
+        use crate::storage::storage_api::ecstore_disk::{DiskAPI, ReadOptions};
+        let set = fixture
+            .env
+            .ecstore
+            .all_set_disks()
+            .into_iter()
+            .next()
+            .expect("target erasure set");
+        let disk = set.disks.read().await.iter().find_map(Clone::clone).expect("local target");
+        let mut fi = rustfs_filemeta::FileInfo::new("destination", 1, 0);
+        fi.erasure.index = 1;
+        fi.version_id = Some(Uuid::new_v4());
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fi.size = 17;
+        fi.parts = vec![rustfs_filemeta::ObjectPartInfo {
+            number: 1,
+            size: 17,
+            actual_size: 17,
+            ..Default::default()
+        }];
+        fi.data = Some(Bytes::from_static(b"target-rpc-inline"));
+        fi.set_inline_data();
+        disk.make_volume("target-rpc").await.expect("target volume");
+        disk.write_metadata("target-rpc", "target-rpc", "staged", fi.clone())
+            .await
+            .expect("stage real inline body");
+        let read = disk
+            .read_version(
+                "target-rpc",
+                "target-rpc",
+                "staged",
+                &fi.version_id.expect("version").to_string(),
+                &ReadOptions {
+                    read_data: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("read staged body before mutation");
+        assert_eq!(read.data, fi.data);
+        let before = tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta"))
+            .await
+            .expect("staged bytes");
+        (disk, fi, before)
+    }
+
+    fn target_rename_request(disk: &super::DiskStore, fi: &rustfs_filemeta::FileInfo) -> Request<RenameDataRequest> {
+        let mut request = Request::new(RenameDataRequest {
+            disk: disk.endpoint().to_string(),
+            src_volume: "target-rpc".to_string(),
+            src_path: "staged".to_string(),
+            dst_volume: "target-rpc".to_string(),
+            dst_path: "destination".to_string(),
+            file_info: serde_json::to_string(fi).expect("real FileInfo JSON"),
+            ..Default::default()
+        });
+        let body = rustfs_protos::canonical_rename_data_request_body(request.get_ref()).expect("canonical target body");
+        set_tonic_canonical_body_digest(&mut request, &body).expect("body digest");
+        // Direct-handler precondition only; this does not stand in for wire authentication.
+        mark_v2_authenticated(&mut request);
+        request
+    }
+
+    #[tokio::test]
+    async fn target_slot_rejects_mismatched_and_repeated_install_before_global_publication() {
+        let fixture = target_rpc_fixture().await;
+        assert!(
+            crate::runtime_sources::current_app_context().is_none(),
+            "requires a separate nextest process"
+        );
+        let wrong = super::ServerContextSlot::with_instance_context(crate::storage::storage_api::new_instance_ctx());
+        let error = crate::runtime_sources::AppContext::ensure_startup_after_iam(
+            fixture.env.ecstore.clone(),
+            Arc::new(KmsServiceManager::new()),
+            &wrong,
+            fixture.iam.clone(),
+        )
+        .expect_err("mismatched startup must fail");
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
+        assert!(wrong.installed_app_context().is_none());
+        assert!(
+            crate::runtime_sources::current_app_context().is_none(),
+            "failed install must not publish globally"
+        );
+        assert!(!wrong.install(fixture.context.clone()), "bool adapter cannot bypass identity checks");
+        let slot = super::ServerContextSlot::with_instance_context(fixture.instance.clone());
+        crate::runtime_sources::AppContext::ensure_startup_after_iam(
+            fixture.env.ecstore.clone(),
+            Arc::new(KmsServiceManager::new()),
+            &slot,
+            fixture.iam.clone(),
+        )
+        .expect("matching startup installation");
+        let installed = slot.installed_app_context().expect("installed A");
+        assert!(Arc::ptr_eq(
+            &crate::runtime_sources::current_app_context().expect("published A"),
+            &installed
+        ));
+        assert_eq!(
+            slot.try_install(installed.clone())
+                .expect_err("same Arc is still a duplicate")
+                .kind(),
+            std::io::ErrorKind::AlreadyExists
+        );
+        assert!(!slot.install(installed.clone()));
+        assert!(Arc::ptr_eq(&slot.installed_app_context().expect("first winner retained"), &installed));
+    }
+
+    #[tokio::test]
+    async fn target_slot_captures_bootstrap_once_and_next_request_observes_ready() {
+        let fixture = target_rpc_fixture().await;
+        let (disk, fi, before) = stage_target_rpc(&fixture).await;
+        let slot = super::ServerContextSlot::with_instance_context(fixture.instance.clone());
+        let service = super::make_server_for_slot(slot.clone());
+        let captured = service.local_mutation_target();
+        slot.try_install(fixture.context.clone())
+            .expect("install after the request captures bootstrap");
+        let super::LocalMutationTarget::Bootstrap(target) = captured else {
+            panic!("pre-install request must capture bootstrap");
+        };
+        assert!(
+            target
+                .rename_local_data(
+                    &disk.endpoint().to_string(),
+                    ("target-rpc", "staged"),
+                    &fi,
+                    ("target-rpc", "destination"),
+                    None
+                )
+                .await
+                .is_err(),
+            "captured request cannot acquire Ready privileges"
+        );
+        assert_eq!(
+            tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta"))
+                .await
+                .expect("original source"),
+            before
+        );
+        assert!(!disk.path().join("target-rpc/destination").exists());
+        assert!(
+            matches!(service.local_mutation_target(), super::LocalMutationTarget::Ready(_)),
+            "the same service must read the installed slot for its next request"
+        );
+        let result = service
+            .rename_data(target_rename_request(&disk, &fi))
+            .await
+            .expect("ready handler")
+            .into_inner();
+        assert!(result.success, "{:?}", result.error);
+    }
+
+    #[tokio::test]
+    async fn target_unbound_slot_never_mutates_a_published_global_store() {
+        let fixture = target_rpc_fixture().await;
+        let (disk, fi, before) = stage_target_rpc(&fixture).await;
+        let published = crate::runtime_sources::publish_test_app_context(fixture.context.clone());
+        assert!(Arc::ptr_eq(&published, &fixture.context));
+        let service = super::make_server_for_slot(super::ServerContextSlot::new());
+        let result = service
+            .rename_data(target_rename_request(&disk, &fi))
+            .await
+            .expect("handler reply")
+            .into_inner();
+        assert!(!result.success);
+        assert!(result.error.is_some());
+        assert_eq!(
+            tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta"))
+                .await
+                .expect("source remains"),
+            before
+        );
+        assert!(!disk.path().join("target-rpc/destination").exists());
+        assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await);
+    }
+
+    #[tokio::test]
+    async fn target_undo_rejects_force_delete_marker_before_mutation() {
+        let fixture = target_rpc_fixture().await;
+        let (disk, fi, before) = stage_target_rpc(&fixture).await;
+        let service = make_server_for_context(Some(fixture.context.clone()));
+        let opts = crate::storage::storage_api::ecstore_disk::DeleteOptions {
+            undo_write: true,
+            ..Default::default()
+        };
+        let mut request = Request::new(DeleteVersionRequest {
+            disk: disk.endpoint().to_string(),
+            volume: "target-rpc".to_string(),
+            path: "staged".to_string(),
+            file_info: serde_json::to_string(&fi).expect("FileInfo"),
+            opts: serde_json::to_string(&opts).expect("opts"),
+            force_del_marker: true,
+            ..Default::default()
+        });
+        let body = rustfs_protos::canonical_delete_version_request_body(request.get_ref()).expect("canonical undo body");
+        set_tonic_canonical_body_digest(&mut request, &body).expect("body digest");
+        mark_v2_authenticated(&mut request);
+        let result = service.delete_version(request).await.expect("handler reply").into_inner();
+        assert!(!result.success);
+        assert!(result.error.is_some());
+        assert_eq!(
+            tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta"))
+                .await
+                .expect("source remains"),
+            before
+        );
+        assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await);
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn target_handler_cancellation_retains_namespace_through_physical_rename() {
+        use crate::storage::storage_api::{
+            LocalPublicationPause, LocalPublicationStage,
+            ecstore_disk::{DiskAPI, ReadOptions},
+        };
+        let fixture = target_rpc_fixture().await;
+        let (disk, fi, _) = stage_target_rpc(&fixture).await;
+        let slot = super::ServerContextSlot::with_instance_context(fixture.instance.clone());
+        slot.try_install(fixture.context.clone()).expect("ready target");
+        let service = super::make_server_for_slot(slot);
+        let mut pause =
+            LocalPublicationPause::install(&disk, "target-rpc", "destination/xl.meta", LocalPublicationStage::PreparedRename)
+                .expect("install scoped physical pause");
+        let mut handler = Box::pin(service.rename_data(target_rename_request(&disk, &fi)));
+        super::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                result = &mut handler => panic!("handler completed before physical entry: {result:?}"),
+                entered = pause.entered() => entered.expect("physical executor entered"),
+            }
+        })
+        .await
+        .expect("bounded physical entry");
+        drop(handler);
+        assert!(
+            fixture.env.ecstore.scanner_data_usage_publication_blocked().await,
+            "dropping the actual target handler must not release its physical owner"
+        );
+        drop(pause);
+        super::timeout(Duration::from_secs(10), async {
+            while fixture.env.ecstore.scanner_data_usage_publication_blocked().await {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("physical owner must drain");
+        let read = disk
+            .read_version(
+                "target-rpc",
+                "target-rpc",
+                "destination",
+                &fi.version_id.expect("version").to_string(),
+                &ReadOptions {
+                    read_data: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("read real late commit");
+        assert_eq!(read.data, fi.data);
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn target_undo_handler_cancellation_retains_owner_until_backup_restoration() {
+        use crate::storage::storage_api::{
+            LocalPublicationPause, LocalPublicationStage,
+            ecstore_disk::{DeleteOptions, DiskAPI, ReadOptions},
+        };
+        let fixture = target_rpc_fixture().await;
+        let (disk, fi, _) = stage_target_rpc(&fixture).await;
+        let mut old = fi.clone();
+        old.data = Some(Bytes::from_static(b"previous-rpc-body"));
+        assert_eq!(old.data.as_ref().expect("old body").len(), 17);
+        disk.write_metadata("target-rpc", "target-rpc", "destination", old.clone())
+            .await
+            .expect("old actual version");
+        let old_bytes = tokio::fs::read(disk.path().join("target-rpc/destination/xl.meta"))
+            .await
+            .expect("old metadata bytes");
+        let committed = fixture
+            .env
+            .ecstore
+            .rename_local_data(
+                &disk.endpoint().to_string(),
+                ("target-rpc", "staged"),
+                &fi,
+                ("target-rpc", "destination"),
+                None,
+            )
+            .await
+            .expect("real overwrite creates rollback backup");
+        let opts = DeleteOptions {
+            undo_write: true,
+            old_data_dir: Some(committed.rollback_data_dir.expect("real rollback backup")),
+            ..Default::default()
+        };
+        let service = make_server_for_context(Some(fixture.context.clone()));
+        let mut request = Request::new(DeleteVersionRequest {
+            disk: disk.endpoint().to_string(),
+            volume: "target-rpc".to_string(),
+            path: "destination".to_string(),
+            file_info: serde_json::to_string(&fi).expect("FileInfo"),
+            opts: serde_json::to_string(&opts).expect("undo options"),
+            ..Default::default()
+        });
+        let body = rustfs_protos::canonical_delete_version_request_body(request.get_ref()).expect("canonical undo body");
+        set_tonic_canonical_body_digest(&mut request, &body).expect("body digest");
+        mark_v2_authenticated(&mut request);
+        let mut pause = LocalPublicationPause::install(&disk, "target-rpc", "destination/xl.meta", LocalPublicationStage::Rename)
+            .expect("pause actual backup restoration");
+        let mut handler = Box::pin(service.delete_version(request));
+        super::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                result = &mut handler => panic!("undo completed before physical entry: {result:?}"),
+                entered = pause.entered() => entered.expect("physical restore entered"),
+            }
+        })
+        .await
+        .expect("bounded physical restore entry");
+        drop(handler);
+        assert!(fixture.env.ecstore.scanner_data_usage_publication_blocked().await);
+        drop(pause);
+        super::timeout(Duration::from_secs(10), async {
+            while fixture.env.ecstore.scanner_data_usage_publication_blocked().await {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("restore owner drains");
+        assert_eq!(
+            tokio::fs::read(disk.path().join("target-rpc/destination/xl.meta"))
+                .await
+                .expect("restored bytes"),
+            old_bytes
+        );
+        let read = disk
+            .read_version(
+                "target-rpc",
+                "target-rpc",
+                "destination",
+                &fi.version_id.expect("version").to_string(),
+                &ReadOptions {
+                    read_data: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("restored readable version");
+        assert_eq!(read.data, old.data);
+    }
+
+    #[tokio::test]
+    async fn rename_data_same_uuid_uses_captured_instance_instead_of_global_disk() {
+        use crate::storage::storage_api::{
+            ECStore,
+            ecstore_disk::{DiskAPI, RUSTFS_META_BUCKET, ReadOptions},
+            init_local_disks_with_instance_ctx, new_instance_ctx, read_config_no_lock,
+        };
+        use rustfs_filemeta::{FileInfo, ObjectPartInfo};
+        use tokio_util::sync::CancellationToken;
+
+        async fn build_store(root: &std::path::Path) -> Arc<ECStore> {
+            let mut endpoints = Vec::new();
+            for index in 0..4 {
+                let path = root.join(format!("disk{index}"));
+                tokio::fs::create_dir_all(&path).await.expect("create instance disk");
+                let mut endpoint = Endpoint::try_from(path.to_str().expect("UTF-8 disk path")).expect("local endpoint");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(index);
+                endpoints.push(endpoint);
+            }
+            let pools = EndpointServerPools(vec![PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 4,
+                endpoints: Endpoints::from(endpoints),
+                cmd_line: "namespace-target-context".to_string(),
+                platform: "test".to_string(),
+            }]);
+            let instance = new_instance_ctx();
+            init_local_disks_with_instance_ctx(&instance, pools.clone())
+                .await
+                .expect("register this instance's real disks");
+            // Match the isolated ECStore fixtures: startup still runs, while
+            // unrelated background recovery is cancelled for this process.
+            let shutdown = CancellationToken::new();
+            shutdown.cancel();
+            ECStore::new_with_instance_ctx("127.0.0.1:0".parse().expect("local address"), pools, shutdown, instance)
+                .await
+                .expect("initialize isolated ECStore")
+        }
+
+        async fn context(store: &Arc<ECStore>) -> Arc<crate::runtime_sources::AppContext> {
+            ObjectStore::new(store.clone())
+                .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
+                .await
+                .expect("seed isolated IAM format");
+            let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("build isolated IAM");
+            Arc::new(crate::runtime_sources::AppContext::with_default_interfaces(
+                store.clone(),
+                iam,
+                Arc::new(KmsServiceManager::new()),
+            ))
+        }
+
+        async fn internal_snapshot(root: &std::path::Path) -> std::collections::BTreeMap<std::path::PathBuf, Option<Vec<u8>>> {
+            let mut snapshot = std::collections::BTreeMap::new();
+            let mut directories = (0..4)
+                .map(|index| std::path::PathBuf::from(format!("disk{index}/{RUSTFS_META_BUCKET}")))
+                .collect::<Vec<_>>();
+            while let Some(relative) = directories.pop() {
+                let mut entries = tokio::fs::read_dir(root.join(&relative))
+                    .await
+                    .expect("read internal snapshot directory");
+                snapshot.insert(relative.clone(), None);
+                while let Some(entry) = entries.next_entry().await.expect("read internal snapshot entry") {
+                    let path = relative.join(entry.file_name());
+                    let file_type = entry.file_type().await.expect("read internal snapshot entry type");
+                    if file_type.is_dir() {
+                        directories.push(path);
+                    } else {
+                        assert!(file_type.is_file(), "fixture snapshot must contain only directories and regular files");
+                        snapshot.insert(path, Some(tokio::fs::read(entry.path()).await.expect("read snapshot file bytes")));
+                    }
+                }
+            }
+            snapshot
+        }
+
+        fn file_info(object: &str, version: Uuid, body: Bytes) -> FileInfo {
+            let mut fi = FileInfo::new(object, 1, 0);
+            fi.erasure.index = 1;
+            fi.name = object.to_string();
+            fi.version_id = Some(version);
+            fi.size = i64::try_from(body.len()).expect("small fixture body");
+            fi.parts = vec![ObjectPartInfo {
+                number: 1,
+                size: body.len(),
+                actual_size: fi.size,
+                ..Default::default()
+            }];
+            fi.data = Some(body);
+            fi.set_inline_data();
+            fi.mod_time = Some(OffsetDateTime::now_utc());
+            fi
+        }
+
+        // Both global publications are first-writer-wins. Run this fixture in
+        // its own nextest process; do not reset or replace another test's state.
+        assert!(
+            crate::runtime_sources::current_app_context().is_none(),
+            "requires an unpublished AppContext"
+        );
+        super::timeout(Duration::from_secs(90), async {
+            let root_b = tempfile::tempdir().expect("instance B directory");
+            let root_a = tempfile::tempdir().expect("instance A directory");
+            let store_b = build_store(root_b.path()).await;
+            let context_b = context(&store_b).await;
+            let published = crate::runtime_sources::publish_test_app_context(context_b.clone());
+            assert!(Arc::ptr_eq(&published, &context_b), "B must win the process AppContext publication");
+
+            // Existing formats require their committed pool metadata on restart.
+            // Copy the complete internal trees, including erasure part data,
+            // without editing disk IDs, cluster identity, epochs or pool topology.
+            super::timeout(Duration::from_secs(10), async {
+                while store_b.scanner_data_usage_publication_blocked().await {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("B startup namespace commits must drain before its snapshot");
+            let snapshot_generation = store_b.scanner_namespace_mutation_generation();
+            let pool_config = read_config_no_lock(store_b.clone(), "pool.bin")
+                .await
+                .expect("read B's actually committed pool metadata");
+            let pool_identity = read_config_no_lock(store_b.clone(), "pool.bin.identity")
+                .await
+                .expect("read B's actually committed pool identity");
+            let snapshot = internal_snapshot(root_b.path()).await;
+            for (relative, contents) in &snapshot {
+                let target = root_a.path().join(relative);
+                match contents {
+                    None => tokio::fs::create_dir_all(target).await.expect("copy internal directory"),
+                    Some(bytes) => tokio::fs::write(target, bytes).await.expect("copy complete internal file"),
+                }
+            }
+            assert_eq!(internal_snapshot(root_a.path()).await, snapshot, "A must receive the complete physical snapshot");
+            assert_eq!(internal_snapshot(root_b.path()).await, snapshot, "B's source snapshot must remain unchanged");
+            assert!(!store_b.scanner_data_usage_publication_blocked().await);
+            assert_eq!(store_b.scanner_namespace_mutation_generation(), snapshot_generation);
+            let store_a = build_store(root_a.path()).await;
+            assert_eq!(
+                read_config_no_lock(store_a.clone(), "pool.bin").await.expect("read A's restarted pool metadata"),
+                pool_config,
+                "A must load the same committed topology without a bootstrap rewrite"
+            );
+            assert_eq!(
+                read_config_no_lock(store_a.clone(), "pool.bin.identity")
+                    .await
+                    .expect("read A's restarted pool identity"),
+                pool_identity,
+                "A must preserve the initialized cluster identity and epoch"
+            );
+            let service = make_server_for_context(Some(context(&store_a).await));
+            assert!(Arc::ptr_eq(&service.resolve_object_store().expect("captured store"), &store_a));
+            assert!(Arc::ptr_eq(
+                &crate::runtime_sources::current_object_store_handle().expect("global store"),
+                &store_b
+            ));
+            let disk_a = store_a.disk_map[&0][0].as_ref().expect("A disk zero").clone();
+            let disk_b = store_b.disk_map[&0][0].as_ref().expect("B disk zero").clone();
+            assert!(disk_a.is_local() && disk_b.is_local());
+            assert!(!Arc::ptr_eq(&disk_a, &disk_b));
+            let disk_id = disk_a.get_disk_id().await.expect("A disk ID").expect("formatted A disk");
+            assert!(!disk_id.is_nil());
+            assert_eq!(disk_b.get_disk_id().await.expect("B disk ID"), Some(disk_id));
+            let global_disk = super::find_local_disk_by_ref(&disk_id.to_string())
+                .await
+                .expect("global UUID lookup must resolve B before the request");
+            assert!(Arc::ptr_eq(&global_disk, &disk_b));
+
+            let volume = "namespace-target-context";
+            let object = "destination";
+            let staging = "staged";
+            let version = Uuid::new_v4();
+            let new_body = Bytes::from_static(b"committed-through-captured-A");
+            let new_fi = file_info(object, version, new_body.clone());
+            let opts = ReadOptions { read_data: true, ..Default::default() };
+            for (disk, old_body) in [
+                (&disk_a, Bytes::from_static(b"old-body-A")),
+                (&disk_b, Bytes::from_static(b"old-body-B")),
+            ] {
+                disk.make_volume(volume).await.expect("create destination volume");
+                disk.write_metadata(volume, volume, object, file_info(object, version, old_body.clone()))
+                    .await
+                    .expect("write real old object metadata and inline body");
+                disk.write_metadata(volume, volume, staging, new_fi.clone())
+                    .await
+                    .expect("stage identical valid metadata on both physical disks");
+                let seeded = disk
+                    .read_version(volume, volume, object, &version.to_string(), &opts)
+                    .await
+                    .expect("decode seeded inline object before invoking the handler");
+                assert_eq!(seeded.data, Some(old_body), "the real reader must return the seeded body");
+            }
+            let a_meta = disk_a.path().join(volume).join(object).join("xl.meta");
+            let b_meta = disk_b.path().join(volume).join(object).join("xl.meta");
+            let a_staging = disk_a.path().join(volume).join(staging).join("xl.meta");
+            let b_staging = disk_b.path().join(volume).join(staging).join("xl.meta");
+            let a_before = tokio::fs::read(&a_meta).await.expect("A old metadata bytes");
+            let b_before = tokio::fs::read(&b_meta).await.expect("B old metadata bytes");
+            let b_staging_before = tokio::fs::read(&b_staging).await.expect("B staged metadata bytes");
+            assert!(tokio::fs::try_exists(&a_staging).await.expect("A staging exists"));
+            assert_ne!(a_before, b_before, "the old on-disk bodies must distinguish A from B");
+            super::timeout(Duration::from_secs(10), async {
+                while store_a.scanner_data_usage_publication_blocked().await
+                    || store_b.scanner_data_usage_publication_blocked().await
+                {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("startup namespace commits must drain before measuring the handler");
+            let generation_before = (
+                store_a.scanner_namespace_mutation_generation(),
+                store_b.scanner_namespace_mutation_generation(),
+            );
+
+            let mut request = Request::new(RenameDataRequest {
+                disk: disk_id.to_string(),
+                src_volume: volume.to_string(),
+                src_path: staging.to_string(),
+                dst_volume: volume.to_string(),
+                dst_path: object.to_string(),
+                file_info: serde_json::to_string(&new_fi).expect("encode real FileInfo"),
+                file_info_bin: Vec::new().into(),
+                scanner_publication_lease_token: Vec::new().into(),
+            });
+            let body = rustfs_protos::canonical_rename_data_request_body(request.get_ref()).expect("canonical rename body");
+            set_tonic_canonical_body_digest(&mut request, &body).expect("body-bound handler request");
+            mark_v2_authenticated(&mut request);
+            let response = super::timeout(Duration::from_secs(10), service.rename_data(request))
+                .await
+                .expect("real rename handler must finish within ten seconds")
+                .expect("rename handler response")
+                .into_inner();
+            assert!(response.success, "the valid staged rename must execute: {:?}", response.error);
+
+            let a_after = disk_a
+                .read_version(volume, volume, object, &version.to_string(), &opts)
+                .await
+                .expect("read A's physical object after rename");
+            let b_after = disk_b
+                .read_version(volume, volume, object, &version.to_string(), &opts)
+                .await
+                .expect("read B's physical object after rename");
+            let a_bytes_after = tokio::fs::read(&a_meta).await.expect("A metadata after rename");
+            let b_bytes_after = tokio::fs::read(&b_meta).await.expect("B metadata after rename");
+            let b_staging_after = tokio::fs::read(&b_staging).await.ok();
+            let generation_after = (
+                store_a.scanner_namespace_mutation_generation(),
+                store_b.scanner_namespace_mutation_generation(),
+            );
+            let pending_after = (
+                store_a.scanner_data_usage_publication_blocked().await,
+                store_b.scanner_data_usage_publication_blocked().await,
+            );
+            for disk in store_a.disk_map.values().chain(store_b.disk_map.values()).flatten().flatten() {
+                disk.close().await.expect("close real fixture disk before assertions and directory cleanup");
+            }
+            assert_eq!(
+                a_after.data,
+                Some(new_body),
+                "RenameData must commit to captured A, even when global B owns the same UUID; B body={:?}, generations={generation_before:?}->{generation_after:?}, pending={pending_after:?}",
+                b_after.data
+            );
+            assert_ne!(a_bytes_after, a_before, "A metadata must actually be replaced");
+            assert_eq!(b_after.data, Some(Bytes::from_static(b"old-body-B")), "B body must remain unchanged");
+            assert_eq!(b_bytes_after, b_before, "B metadata must remain byte-for-byte unchanged");
+            assert_eq!(b_staging_after, Some(b_staging_before), "B staging must not be consumed");
+            assert_eq!(pending_after, (false, false), "both stores must reach a stable terminal state");
+        })
+        .await
+        .expect("two-instance handler fixture must finish within ninety seconds");
+    }
+
     #[tokio::test]
     async fn test_make_volumes_invalid_disk() {
         let service = create_test_node_service();

rustfs/src/storage/rpc/node_service/disk.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::NodeService;
+use super::{LocalMutationTarget, NodeService};
 use crate::storage::storage_api::rpc_consumer::node_service::{
     BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq,
     ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count,
@@ -39,6 +39,69 @@ use tonic::{Request, Response, Status};
 use tracing::debug;
 use uuid::Uuid;
 
+#[cfg(feature = "e2e-test-hooks")]
+fn startup_cas_rename_observation(
+    target: &LocalMutationTarget,
+    request: &RenameDataRequest,
+    file_info: &FileInfo,
+) -> Option<serde_json::Value> {
+    use sha2::{Digest, Sha256};
+    if request.dst_volume != ".rustfs.sys" || !matches!(request.dst_path.as_str(), "pool.bin" | "pool.bin.identity") {
+        return None;
+    }
+    let nonce = uuid::Uuid::parse_str(&std::env::var("RUSTFS_E2E_STARTUP_CAS_NONCE").ok()?).ok()?;
+    let body = rustfs_protos::canonical_rename_data_request_body(request).ok()?;
+    Some(serde_json::json!({
+        "kind": "receiver", "nonce": nonce, "pid": std::process::id(),
+        "target": match target { LocalMutationTarget::Ready(_) => "ready", LocalMutationTarget::Bootstrap(_) => "bootstrap", LocalMutationTarget::Unbound => "unbound" },
+        "disk": request.disk, "src_volume": request.src_volume, "src_path": request.src_path,
+        "dst_volume": request.dst_volume, "dst_path": request.dst_path,
+        "body_sha256": rustfs_utils::crypto::hex(Sha256::digest(body)),
+        "etag": file_info.metadata.get("etag"),
+        "mod_time": file_info.mod_time.map(|time| time.unix_timestamp_nanos().to_string()),
+    }))
+}
+
+impl LocalMutationTarget {
+    async fn rename_local_data(
+        &self,
+        disk_ref: &str,
+        source: (&str, &str),
+        fi: &FileInfo,
+        destination: (&str, &str),
+        scanner_token: Option<Uuid>,
+    ) -> Result<RenameDataResp, DiskError> {
+        match self {
+            Self::Ready(store) => {
+                store
+                    .rename_local_data(disk_ref, source, fi, destination, scanner_token)
+                    .await
+            }
+            Self::Bootstrap(target) => {
+                target
+                    .rename_local_data(disk_ref, source, fi, destination, scanner_token)
+                    .await
+            }
+            Self::Unbound => Err(DiskError::other("target disk instance is unavailable")),
+        }
+    }
+
+    async fn undo_local_write(
+        &self,
+        disk_ref: &str,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        opts: DeleteOptions,
+    ) -> Result<(), DiskError> {
+        match self {
+            Self::Ready(store) => store.undo_local_write(disk_ref, volume, path, fi, opts).await,
+            Self::Bootstrap(target) => target.undo_local_write(disk_ref, volume, path, fi, opts).await,
+            Self::Unbound => Err(DiskError::other("target disk instance is unavailable")),
+        }
+    }
+}
+
 /// Initial capacity hint (bytes) for typical small msgpack requests and responses.
 const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
 const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024;
@@ -670,55 +733,59 @@ impl NodeService {
             "delete_version",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
-            let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
-                Ok(file_info) => file_info,
-                Err(err) => {
-                    return Ok(Response::new(DeleteVersionResponse {
-                        success: false,
-                        raw_file_info: "".to_string(),
-                        error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()),
-                    }));
-                }
-            };
-            let opts = match decode_msgpack_or_json::<DeleteOptions>(&request.opts_bin, &request.opts, "DeleteOptions") {
-                Ok(opts) => opts,
-                Err(err) => {
-                    return Ok(Response::new(DeleteVersionResponse {
-                        success: false,
-                        raw_file_info: "".to_string(),
-                        error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
-                    }));
-                }
-            };
-            match disk
-                .delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts)
+        let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
+            Ok(file_info) => file_info,
+            Err(err) => {
+                return Ok(Response::new(DeleteVersionResponse {
+                    success: false,
+                    raw_file_info: "".to_string(),
+                    error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()),
+                }));
+            }
+        };
+        let opts = match decode_msgpack_or_json::<DeleteOptions>(&request.opts_bin, &request.opts, "DeleteOptions") {
+            Ok(opts) => opts,
+            Err(err) => {
+                return Ok(Response::new(DeleteVersionResponse {
+                    success: false,
+                    raw_file_info: "".to_string(),
+                    error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
+                }));
+            }
+        };
+        let result = if opts.undo_write {
+            if request.force_del_marker {
+                Err(DiskError::other("undo_write cannot force a delete marker"))
+            } else {
+                let target = self.local_mutation_target();
+                target
+                    .undo_local_write(&request.disk, &request.volume, &request.path, file_info, opts)
+                    .await
+            }
+        } else if let Some(disk) = self.find_disk(&request.disk).await {
+            disk.delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts)
                 .await
-            {
-                Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) {
-                    Ok(raw_file_info) => Ok(Response::new(DeleteVersionResponse {
-                        success: true,
-                        raw_file_info,
-                        error: None,
-                    })),
-                    Err(err) => Ok(Response::new(DeleteVersionResponse {
-                        success: false,
-                        raw_file_info: "".to_string(),
-                        error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
-                    })),
-                },
+        } else {
+            Err(DiskError::other("cannot find disk"))
+        };
+        match result {
+            Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) {
+                Ok(raw_file_info) => Ok(Response::new(DeleteVersionResponse {
+                    success: true,
+                    raw_file_info,
+                    error: None,
+                })),
                 Err(err) => Ok(Response::new(DeleteVersionResponse {
                     success: false,
                     raw_file_info: "".to_string(),
-                    error: Some(err.into()),
+                    error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                 })),
-            }
-        } else {
-            Ok(Response::new(DeleteVersionResponse {
+            },
+            Err(err) => Ok(Response::new(DeleteVersionResponse {
                 success: false,
                 raw_file_info: "".to_string(),
-                error: Some(DiskError::other("cannot find disk".to_string()).into()),
-            }))
+                error: Some(err.into()),
+            })),
         }
     }
 
@@ -1206,98 +1273,70 @@ impl NodeService {
             "rename_data",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
-            let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) {
-                Ok(file_info) => file_info,
-                Err(err) => {
-                    return Ok(Response::new(RenameDataResponse {
-                        success: false,
-                        rename_data_resp: String::new(),
-                        rename_data_resp_bin: Vec::new().into(),
-                        error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()),
-                    }));
-                }
-            };
-            let scanner_publication_lease_token = if request.scanner_publication_lease_token.is_empty() {
-                None
-            } else {
-                let token = Uuid::from_slice(&request.scanner_publication_lease_token)
-                    .map_err(|_| Status::invalid_argument("scanner publication lease token must be a UUID"))?;
-                if token.is_nil() {
-                    return Err(Status::invalid_argument("scanner publication lease token must not be nil"));
-                }
-                Some(token)
-            };
-            // The target owns this read guard.  It must span the complete
-            // disk rename, not merely the preflight, so a movement transition
-            // cannot restart after validation and before rename linearization.
-            let scanner_publication_lease_guard: Option<Arc<dyn Send + Sync>> =
-                if let Some(token) = scanner_publication_lease_token {
-                    let Some(store) = self.resolve_object_store() else {
-                        return Ok(Response::new(RenameDataResponse {
-                            success: false,
-                            rename_data_resp: String::new(),
-                            rename_data_resp_bin: Vec::new().into(),
-                            error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
-                        }));
-                    };
-                    match store.acquire_scanner_publication_lease_guard(token).await {
-                        Ok(guard) => Some(Arc::new(guard)),
-                        Err(err) => {
-                            return Ok(Response::new(RenameDataResponse {
-                                success: false,
-                                rename_data_resp: String::new(),
-                                rename_data_resp_bin: Vec::new().into(),
-                                error: Some(DiskError::other(err.to_string()).into()),
-                            }));
-                        }
-                    }
-                } else {
-                    None
-                };
-            let request_decoded_from_msgpack = decoded_file_info.from_msgpack;
-            match disk
-                .rename_data_borrowed_with_fence_and_guard(
-                    &request.src_volume,
-                    &request.src_path,
-                    &decoded_file_info.value,
-                    &request.dst_volume,
-                    &request.dst_path,
-                    scanner_publication_lease_token,
-                    scanner_publication_lease_guard,
-                )
-                .await
-            {
-                Ok(rename_data_resp) => {
-                    match encode_rename_data_response_payloads(&rename_data_resp, request_decoded_from_msgpack) {
-                        Ok((rename_data_resp, rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse {
-                            success: true,
-                            rename_data_resp,
-                            rename_data_resp_bin: rename_data_resp_bin.into(),
-                            error: None,
-                        })),
-                        Err(err) => Ok(Response::new(RenameDataResponse {
-                            success: false,
-                            rename_data_resp: String::new(),
-                            rename_data_resp_bin: Vec::new().into(),
-                            error: Some(err.into()),
-                        })),
-                    }
-                }
+        let target = self.local_mutation_target();
+        #[cfg(feature = "e2e-test-hooks")]
+        super::rename_target_capture_test_hook::wait(&target, &request).await;
+        let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) {
+            Ok(file_info) => file_info,
+            Err(err) => {
+                return Ok(Response::new(RenameDataResponse {
+                    success: false,
+                    rename_data_resp: String::new(),
+                    rename_data_resp_bin: Vec::new().into(),
+                    error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()),
+                }));
+            }
+        };
+        let scanner_publication_lease_token = if request.scanner_publication_lease_token.is_empty() {
+            None
+        } else {
+            let token = Uuid::from_slice(&request.scanner_publication_lease_token)
+                .map_err(|_| Status::invalid_argument("scanner publication lease token must be a UUID"))?;
+            if token.is_nil() {
+                return Err(Status::invalid_argument("scanner publication lease token must not be nil"));
+            }
+            Some(token)
+        };
+        let request_decoded_from_msgpack = decoded_file_info.from_msgpack;
+        #[cfg(feature = "e2e-test-hooks")]
+        let observation = startup_cas_rename_observation(&target, &request, &decoded_file_info.value);
+        let result = target
+            .rename_local_data(
+                &request.disk,
+                (&request.src_volume, &request.src_path),
+                &decoded_file_info.value,
+                (&request.dst_volume, &request.dst_path),
+                scanner_publication_lease_token,
+            )
+            .await;
+        #[cfg(feature = "e2e-test-hooks")]
+        if let Some(mut observation) = observation {
+            observation["ok"] = serde_json::json!(result.is_ok());
+            observation["error"] = serde_json::json!(result.as_ref().err().map(ToString::to_string));
+            let line = format!("RUSTFS_E2E_STARTUP_CAS {observation}\n");
+            let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes());
+        }
+        match result {
+            Ok(rename_data_resp) => match encode_rename_data_response_payloads(&rename_data_resp, request_decoded_from_msgpack) {
+                Ok((rename_data_resp, rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse {
+                    success: true,
+                    rename_data_resp,
+                    rename_data_resp_bin: rename_data_resp_bin.into(),
+                    error: None,
+                })),
                 Err(err) => Ok(Response::new(RenameDataResponse {
                     success: false,
                     rename_data_resp: String::new(),
                     rename_data_resp_bin: Vec::new().into(),
                     error: Some(err.into()),
                 })),
-            }
-        } else {
-            Ok(Response::new(RenameDataResponse {
+            },
+            Err(err) => Ok(Response::new(RenameDataResponse {
                 success: false,
                 rename_data_resp: String::new(),
                 rename_data_resp_bin: Vec::new().into(),
-                error: Some(DiskError::other("cannot find disk".to_string()).into()),
-            }))
+                error: Some(err.into()),
+            })),
         }
     }
 

rustfs/src/storage/storage_api.rs

@@ -377,10 +377,12 @@ pub(crate) mod timeout_wrapper_consumer {
 }
 
 pub(crate) mod tonic_service_consumer {
+    #[cfg(test)]
+    pub(crate) use super::super::tonic_service::make_server;
     #[cfg(test)]
     pub(crate) use super::super::tonic_service::{heal_topology_fingerprint, make_heal_control_server_for_source};
     pub(crate) use super::super::tonic_service::{
-        make_heal_control_server_with_cache, make_scanner_control_server, make_server, make_tier_mutation_control_server,
+        make_heal_control_server_with_cache, make_scanner_control_server, make_server_for_slot, make_tier_mutation_control_server,
     };
 }
 
@@ -600,8 +602,8 @@ pub(crate) mod ecstore_storage {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
     pub(crate) use rustfs_ecstore::api::storage::{
-        ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, all_local_disk_path,
-        find_local_disk_by_ref, init_local_disks_with_instance_ctx, init_lock_clients,
+        BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
+        all_local_disk_path, find_local_disk_by_ref, init_local_disks_with_instance_ctx, init_lock_clients,
         prewarm_local_disk_id_map_with_instance_ctx,
     };
 }
@@ -677,6 +679,9 @@ type EcstoreReplicationStats = ecstore_bucket::replication::ReplicationStats;
 pub(crate) type DynReplicationPool = StorageReplicationPoolHandle;
 pub(crate) type DynReader = ecstore_rio::DynReader;
 pub(crate) type ECStore = ecstore_storage::ECStore;
+pub(crate) type BootstrapLocalTarget = ecstore_storage::BootstrapLocalTarget;
+#[cfg(all(test, not(windows)))]
+pub(crate) use rustfs_ecstore::api::disk::{LocalPublicationPause, LocalPublicationStage};
 pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
 #[cfg(test)]
 pub(crate) type Endpoints = ecstore_layout::Endpoints;

rustfs/src/storage/tonic_service.rs

@@ -13,8 +13,14 @@
 // limitations under the License.
 
 pub(crate) use crate::storage::rpc::node_service::make_heal_control_server_with_cache;
-pub(crate) use crate::storage::rpc::node_service::make_scanner_control_server;
 #[cfg(test)]
 pub(crate) use crate::storage::rpc::node_service::{heal::heal_topology_fingerprint, make_heal_control_server_for_source};
+pub(crate) use crate::storage::rpc::node_service::{make_scanner_control_server, make_server_for_slot};
 pub use crate::storage::rpc::{make_heal_control_server, make_server, make_tier_mutation_control_server};
 pub type NodeService = crate::storage::rpc::NodeService;
+
+#[cfg(feature = "e2e-test-hooks")]
+#[doc(hidden)]
+pub use crate::storage::rpc::node_service::rename_target_capture_test_hook::{
+    RenameTargetCapturePause, pause_rename_after_target_capture,
+};

rustfs/src/storage_api.rs

@@ -171,12 +171,15 @@ pub(crate) mod server {
         }
 
         pub(crate) mod tonic_service {
+            #[cfg(test)]
+            pub(crate) use crate::storage::storage_api::tonic_service_consumer::make_server;
             #[cfg(test)]
             pub(crate) use crate::storage::storage_api::tonic_service_consumer::{
                 heal_topology_fingerprint, make_heal_control_server_for_source,
             };
             pub(crate) use crate::storage::storage_api::tonic_service_consumer::{
-                make_heal_control_server_with_cache, make_scanner_control_server, make_server, make_tier_mutation_control_server,
+                make_heal_control_server_with_cache, make_scanner_control_server, make_server_for_slot,
+                make_tier_mutation_control_server,
             };
         }
     }

rustfs/tests/embedded_multi_instance_test.rs

@@ -540,3 +540,530 @@ async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed
     server_b.shutdown().await;
     server_a.shutdown().await;
 }
+
+#[cfg(feature = "e2e-test-hooks")]
+mod signed_target_rpc {
+    use super::{common, find_available_port, pause_embedded_startup_after_http_bind, sha256_hex};
+    use bytes::Bytes;
+    use futures::FutureExt;
+    use hyper_util::rt::TokioIo;
+    use rustfs::app::context::resolve_object_store_handle;
+    use rustfs::embedded::RustFSServerBuilder;
+    use rustfs_ecstore::api::disk::{DiskAPI, DiskError, DiskOption, DiskStore, Endpoint, ReadOptions, new_disk};
+    use rustfs_ecstore::api::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience};
+    use rustfs_filemeta::{FileInfo, ObjectPartInfo};
+    use rustfs_protos::proto_gen::node_service::{RenameDataRequest, RenameDataResponse, node_service_client::NodeServiceClient};
+    use std::net::SocketAddr;
+    use std::path::Path;
+    use std::sync::{
+        Arc, Mutex,
+        atomic::{AtomicUsize, Ordering},
+    };
+    use std::time::Duration;
+    use time::OffsetDateTime;
+    use tokio::net::TcpStream;
+    use tokio::time::timeout;
+    use tonic::transport::Channel;
+    use uuid::Uuid;
+
+    const WAIT: Duration = Duration::from_secs(30);
+    const INTERNAL_VOLUME: &str = ".rustfs.sys/tmp";
+    const USER_VOLUME: &str = "target-transport";
+
+    struct SingleConnection {
+        client: NodeServiceClient<Channel>,
+        local: SocketAddr,
+        peer: SocketAddr,
+        attempts: Arc<AtomicUsize>,
+    }
+
+    impl SingleConnection {
+        async fn connect(address: SocketAddr) -> Self {
+            let socket = timeout(WAIT, TcpStream::connect(address))
+                .await
+                .expect("bounded real TCP connection")
+                .expect("connect to the production listener");
+            let local = socket.local_addr().expect("client socket identity");
+            let peer = socket.peer_addr().expect("listener socket identity");
+            let socket = Arc::new(Mutex::new(Some(socket)));
+            let attempts = Arc::new(AtomicUsize::new(0));
+            let connector_attempts = attempts.clone();
+            let channel = timeout(
+                WAIT,
+                tonic::transport::Endpoint::from_shared(format!("http://{address}"))
+                    .expect("local endpoint")
+                    .timeout(WAIT)
+                    .connect_with_connector(tower::service_fn(move |_: http::Uri| {
+                        connector_attempts.fetch_add(1, Ordering::SeqCst);
+                        // A channel may reconnect implicitly. This fixture has exactly one
+                        // already-connected socket and fails every subsequent dial attempt.
+                        let socket = socket.lock().expect("single socket lock").take();
+                        async move {
+                            socket.map(TokioIo::new).ok_or_else(|| {
+                                std::io::Error::new(std::io::ErrorKind::ConnectionAborted, "implicit reconnect forbidden")
+                            })
+                        }
+                    })),
+            )
+            .await
+            .expect("bounded HTTP/2 handshake")
+            .expect("HTTP/2 over the original TCP connection");
+            Self {
+                client: NodeServiceClient::new(channel),
+                local,
+                peer,
+                attempts,
+            }
+        }
+
+        fn assert_original_connection(&self) {
+            assert_eq!(self.attempts.load(Ordering::SeqCst), 1, "the channel must not redial");
+        }
+
+        async fn rename(&mut self, request: tonic::Request<RenameDataRequest>) -> RenameDataResponse {
+            let response = timeout(WAIT, self.client.rename_data(request))
+                .await
+                .expect("bounded signed RenameData")
+                .expect("production authentication and RPC routing")
+                .into_inner();
+            self.assert_original_connection();
+            response
+        }
+    }
+
+    async fn local_fixture_disk(root: &Path) -> DiskStore {
+        let mut endpoint = Endpoint::try_from(root.to_str().expect("UTF-8 fixture root")).expect("local disk endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        new_disk(&endpoint, &DiskOption::default())
+            .await
+            .expect("open real fixture disk")
+    }
+
+    async fn stage(disk: &DiskStore, volume: &str, path: &str, body: &'static [u8]) -> FileInfo {
+        match disk.make_volume(volume).await {
+            Ok(()) | Err(DiskError::VolumeExists) => {}
+            Err(err) => panic!("create fixture volume: {err}"),
+        }
+        let mut fi = FileInfo::new(path, 1, 0);
+        fi.erasure.index = 1;
+        fi.version_id = Some(Uuid::new_v4());
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fi.size = i64::try_from(body.len()).expect("small fixture");
+        fi.parts = vec![ObjectPartInfo {
+            number: 1,
+            size: body.len(),
+            actual_size: fi.size,
+            ..Default::default()
+        }];
+        fi.data = Some(Bytes::from_static(body));
+        fi.set_inline_data();
+        disk.write_metadata(volume, volume, path, fi.clone())
+            .await
+            .expect("stage real xl.meta");
+        assert_body(disk, volume, path, &fi).await;
+        fi
+    }
+
+    async fn assert_body(disk: &DiskStore, volume: &str, path: &str, fi: &FileInfo) {
+        let read = disk
+            .read_version(
+                volume,
+                volume,
+                path,
+                &fi.version_id.expect("version").to_string(),
+                &ReadOptions {
+                    read_data: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("decode actual inline object bytes");
+        assert_eq!(read.data, fi.data);
+    }
+
+    fn signed_rename(
+        disk: &DiskStore,
+        volume: &str,
+        source: &str,
+        destination: &str,
+        fi: &FileInfo,
+    ) -> tonic::Request<RenameDataRequest> {
+        let payload = RenameDataRequest {
+            disk: disk.endpoint().to_string(),
+            src_volume: volume.to_owned(),
+            src_path: source.to_owned(),
+            dst_volume: volume.to_owned(),
+            dst_path: destination.to_owned(),
+            file_info: serde_json::to_string(fi).expect("real FileInfo JSON"),
+            ..Default::default()
+        };
+        let canonical = rustfs_protos::canonical_rename_data_request_body(&payload).expect("canonical mutation body");
+        // The current production interceptor uses the process RPC identity. Keep
+        // that authentication contract while testing listener-local disk routing.
+        let identity = rustfs_common::try_get_global_local_node_name().expect("startup published the RPC identity");
+        let audience = normalize_tonic_rpc_audience(&identity).expect("RPC audience");
+        let headers =
+            gen_tonic_signature_headers(&audience, "node_service.NodeService", "RenameData", Some(&sha256_hex(&canonical)))
+                .expect("production v2 signing with the configured shared secret");
+        assert_eq!(headers.get("x-rustfs-rpc-auth-version").expect("v2 metadata"), "2");
+        let mut request = tonic::Request::new(payload);
+        *request.metadata_mut() = tonic::metadata::MetadataMap::from_headers(headers);
+        request
+    }
+
+    #[test]
+    fn signed_target_rpc_uses_listener_instance_across_install_and_reconnect() {
+        common::run_embedded_test(|| async {
+            timeout(WAIT * 6, signed_target_rpc_body())
+                .await
+                .expect("bounded listener/startup/transport fixture");
+        });
+    }
+
+    async fn signed_target_rpc_body() {
+        // B installs the process default first; A must remain a different target
+        // both before and after its own application context is installed.
+        let root_b = tempfile::tempdir().expect("B root");
+        let server_b = timeout(
+            WAIT,
+            RustFSServerBuilder::new()
+                .address(format!("127.0.0.1:{}", find_available_port().expect("B port")))
+                .volume(root_b.path().to_str().expect("B path"))
+                .access_key("target-transport-access")
+                .secret_key("target-transport-secret")
+                .build(),
+        )
+        .await
+        .expect("bounded B startup")
+        .expect("start global B");
+        let global_b = resolve_object_store_handle().expect("B installed process AppContext");
+        let disk_b = local_fixture_disk(root_b.path()).await;
+        let global_endpoints = global_b.instance_endpoints().expect("B instance topology");
+        let global_paths: Vec<_> = global_endpoints
+            .0
+            .iter()
+            .flat_map(|pool| pool.endpoints.as_ref().iter())
+            .map(ToString::to_string)
+            .collect();
+        assert_eq!(
+            global_paths,
+            vec![disk_b.endpoint().to_string()],
+            "the ambient store must really own B's disk"
+        );
+        let sentinel = stage(&disk_b, USER_VOLUME, "sentinel", b"global-B-must-survive").await;
+        let sentinel_path = root_b.path().join(USER_VOLUME).join("sentinel/xl.meta");
+        let sentinel_bytes = tokio::fs::read(&sentinel_path).await.expect("B's committed bytes");
+
+        let root_a = tempfile::tempdir().expect("A root");
+        let port_a = find_available_port().expect("A port");
+        let address_a: SocketAddr = format!("127.0.0.1:{port_a}").parse().expect("A address");
+        let mut barrier = pause_embedded_startup_after_http_bind(port_a);
+        let startup_a = RustFSServerBuilder::new()
+            .address(address_a.to_string())
+            .volume(root_a.path().to_str().expect("A path"))
+            .access_key("target-transport-access")
+            .secret_key("target-transport-secret")
+            .build();
+        tokio::pin!(startup_a);
+        timeout(WAIT, async {
+            tokio::select! {
+                () = barrier.wait_until_http_bound() => {}
+                result = startup_a.as_mut() => {
+                    let _unexpected_server = result.expect("A startup before barrier");
+                    panic!("A must pause after bind and before ECStore/AppContext");
+                }
+            }
+        })
+        .await
+        .expect("bounded A HTTP-bind barrier");
+
+        // Catch assertion failures only to release the real startup barrier and
+        // obtain a shutdown-capable server handle before resuming the failure.
+        let pre_ready = std::panic::AssertUnwindSafe(async {
+            assert!(Arc::ptr_eq(&global_b, &resolve_object_store_handle().expect("global B remains live")));
+            let disk_a = local_fixture_disk(root_a.path()).await;
+            let internal = stage(&disk_a, INTERNAL_VOLUME, "transport-staged", b"pre-ready-internal-body").await;
+            let user = stage(&disk_a, USER_VOLUME, "staged", b"listener-A-user-body").await;
+            let user_before = tokio::fs::read(root_a.path().join(USER_VOLUME).join("staged/xl.meta"))
+                .await
+                .expect("A staged user bytes");
+            let mut connection = SingleConnection::connect(address_a).await;
+            let mut invalid_signature = signed_rename(&disk_a, INTERNAL_VOLUME, "transport-staged", "bad-signature", &internal);
+            invalid_signature
+                .metadata_mut()
+                .insert("x-rustfs-rpc-signature-v2", "00".parse().expect("invalid MAC header"));
+            let status = timeout(WAIT, connection.client.rename_data(invalid_signature))
+                .await
+                .expect("bounded invalid-signature response")
+                .expect_err("production interceptor must reject a bad signature");
+            assert_eq!(status.code(), tonic::Code::Unauthenticated);
+            assert!(!root_a.path().join(INTERNAL_VOLUME).join("bad-signature/xl.meta").exists());
+            assert_body(&disk_a, INTERNAL_VOLUME, "transport-staged", &internal).await;
+
+            let committed = connection
+                .rename(signed_rename(
+                    &disk_a,
+                    INTERNAL_VOLUME,
+                    "transport-staged",
+                    "transport-published",
+                    &internal,
+                ))
+                .await;
+            assert!(
+                committed.success,
+                "Bootstrap must commit internal metadata through the bound A registry: {:?}",
+                committed.error
+            );
+            assert_body(&disk_a, INTERNAL_VOLUME, "transport-published", &internal).await;
+
+            let denied = connection
+                .rename(signed_rename(&disk_a, USER_VOLUME, "staged", "destination", &user))
+                .await;
+            assert!(!denied.success, "Bootstrap must reject a real user mutation");
+            let error: DiskError = denied.error.expect("typed bootstrap rejection").into();
+            assert_eq!(error, DiskError::FileAccessDenied);
+            assert_eq!(
+                tokio::fs::read(root_a.path().join(USER_VOLUME).join("staged/xl.meta"))
+                    .await
+                    .expect("unchanged A source"),
+                user_before
+            );
+            assert!(!root_a.path().join(USER_VOLUME).join("destination/xl.meta").exists());
+
+            assert_eq!(tokio::fs::read(&sentinel_path).await.expect("unchanged B bytes"), sentinel_bytes);
+            assert_body(&disk_b, USER_VOLUME, "sentinel", &sentinel).await;
+            connection.assert_original_connection();
+            (connection, disk_a, user)
+        })
+        .catch_unwind()
+        .await;
+
+        barrier.release();
+        let server_a = timeout(WAIT, startup_a.as_mut())
+            .await
+            .expect("bounded A context installation")
+            .expect("A startup after real internal metadata commit");
+        let (mut connection, disk_a, user) = match pre_ready {
+            Ok(fixture) => fixture,
+            Err(panic) => {
+                timeout(WAIT, server_a.shutdown()).await.expect("bounded A failure cleanup");
+                timeout(WAIT, server_b.shutdown()).await.expect("bounded B failure cleanup");
+                std::panic::resume_unwind(panic);
+            }
+        };
+        assert!(Arc::ptr_eq(
+            &global_b,
+            &resolve_object_store_handle().expect("A install preserves global B")
+        ));
+        assert_eq!(connection.peer, server_a.address());
+        assert_body(&disk_a, USER_VOLUME, "staged", &user).await;
+        let committed = connection
+            .rename(signed_rename(&disk_a, USER_VOLUME, "staged", "destination", &user))
+            .await;
+        assert!(
+            committed.success,
+            "the same accepted connection must observe Ready for its next request: {:?}",
+            committed.error
+        );
+        assert_body(&disk_a, USER_VOLUME, "destination", &user).await;
+
+        // Keep the first connection open so the OS cannot recycle its 4-tuple.
+        let mut reconnected = SingleConnection::connect(address_a).await;
+        assert_ne!(reconnected.local, connection.local);
+        assert_eq!(reconnected.peer, connection.peer);
+        let committed = reconnected
+            .rename(signed_rename(&disk_a, USER_VOLUME, "destination", "reconnected", &user))
+            .await;
+        assert!(committed.success, "new connections must retain listener A: {:?}", committed.error);
+        assert_body(&disk_a, USER_VOLUME, "reconnected", &user).await;
+        assert_eq!(tokio::fs::read(&sentinel_path).await.expect("B remains unchanged"), sentinel_bytes);
+        assert_body(&disk_b, USER_VOLUME, "sentinel", &sentinel).await;
+        assert!(!root_b.path().join(USER_VOLUME).join("reconnected/xl.meta").exists());
+        connection.assert_original_connection();
+        reconnected.assert_original_connection();
+        drop(reconnected);
+        drop(connection);
+        drop(disk_a);
+        drop(disk_b);
+        timeout(WAIT, server_a.shutdown()).await.expect("bounded A shutdown");
+        timeout(WAIT, server_b.shutdown()).await.expect("bounded B shutdown");
+    }
+
+    #[test]
+    fn signed_bootstrap_request_does_not_upgrade_after_context_installation() {
+        common::run_embedded_test(|| async {
+            timeout(WAIT * 6, signed_delayed_bootstrap_body())
+                .await
+                .expect("bounded delayed Bootstrap fixture");
+        });
+    }
+
+    async fn signed_delayed_bootstrap_body() {
+        use rustfs::storage::tonic_service::pause_rename_after_target_capture;
+
+        let root_b = tempfile::tempdir().expect("B root");
+        let server_b = timeout(
+            WAIT,
+            RustFSServerBuilder::new()
+                .address(format!("127.0.0.1:{}", find_available_port().expect("B port")))
+                .volume(root_b.path().to_str().expect("B path"))
+                .access_key("delayed-bootstrap-access")
+                .secret_key("delayed-bootstrap-secret")
+                .build(),
+        )
+        .await
+        .expect("bounded B startup")
+        .expect("start global B");
+        let global_b = resolve_object_store_handle().expect("B's published context");
+        let disk_b = local_fixture_disk(root_b.path()).await;
+        let endpoints = global_b.instance_endpoints().expect("B instance topology");
+        let paths: Vec<_> = endpoints
+            .0
+            .iter()
+            .flat_map(|pool| pool.endpoints.as_ref().iter())
+            .map(ToString::to_string)
+            .collect();
+        assert_eq!(paths, [disk_b.endpoint().to_string()], "the ambient store owns B");
+        stage(&disk_b, USER_VOLUME, "delayed-sentinel", b"B-is-not-the-listener-target").await;
+        let sentinel_path = root_b.path().join(USER_VOLUME).join("delayed-sentinel/xl.meta");
+        let sentinel_before = tokio::fs::read(&sentinel_path).await.expect("B sentinel bytes");
+
+        let root_a = tempfile::tempdir().expect("A root");
+        let port_a = find_available_port().expect("A port");
+        let address_a = format!("127.0.0.1:{port_a}").parse().expect("A address");
+        let mut startup_barrier = Some(pause_embedded_startup_after_http_bind(port_a));
+        let startup_a = RustFSServerBuilder::new()
+            .address(format!("127.0.0.1:{port_a}"))
+            .volume(root_a.path().to_str().expect("A path"))
+            .access_key("delayed-bootstrap-access")
+            .secret_key("delayed-bootstrap-secret")
+            .build();
+        tokio::pin!(startup_a);
+        timeout(WAIT, async {
+            tokio::select! {
+                () = startup_barrier.as_mut().expect("startup barrier").wait_until_http_bound() => {}
+                startup = startup_a.as_mut() => {
+                    let _unexpected_server = startup.expect("A initial startup");
+                    panic!("A must reach its pre-AppContext barrier");
+                }
+            }
+        })
+        .await
+        .expect("bounded A listener startup");
+
+        let disk_a = local_fixture_disk(root_a.path()).await;
+        let delayed_info = stage(&disk_a, USER_VOLUME, "delayed-source", b"captured-Bootstrap-must-not-publish").await;
+        let control_info = stage(&disk_a, USER_VOLUME, "control-source", b"new-Ready-request-can-publish").await;
+        let source_path = root_a.path().join(USER_VOLUME).join("delayed-source/xl.meta");
+        let destination_path = root_a.path().join(USER_VOLUME).join("delayed-destination/xl.meta");
+        let source_before = tokio::fs::read(&source_path).await.expect("delayed source bytes");
+        let mut connection = SingleConnection::connect(address_a).await;
+        let mut server_a = None;
+        let mut startup_finished = false;
+
+        let (observations, delayed_result, source_after, destination_exists, sentinel_after) = {
+            let mut capture =
+                pause_rename_after_target_capture(&disk_a.endpoint().to_string(), USER_VOLUME, "delayed-destination");
+            let mut delayed_client = connection.client.clone();
+            let delayed = delayed_client.rename_data(signed_rename(
+                &disk_a,
+                USER_VOLUME,
+                "delayed-source",
+                "delayed-destination",
+                &delayed_info,
+            ));
+            tokio::pin!(delayed);
+            let mut early_response = None;
+
+            // Bound all work while the request is parked to less than the
+            // existing channel's 30-second deadline; no timeout is disabled.
+            let observations = std::panic::AssertUnwindSafe(timeout(Duration::from_secs(20), async {
+                let was_bootstrap = tokio::select! {
+                    observed = capture.wait_until_captured() => observed,
+                    response = delayed.as_mut() => {
+                        early_response = Some(response);
+                        panic!("signed request finished before the capture pause: {early_response:?}");
+                    },
+                };
+                assert!(was_bootstrap, "the actual authenticated handler captured Bootstrap");
+                assert!(Arc::ptr_eq(&global_b, &resolve_object_store_handle().expect("global B")));
+                assert_eq!(tokio::fs::read(&source_path).await.expect("source before install"), source_before);
+                assert!(!destination_path.exists());
+
+                startup_barrier.take().expect("unreleased startup barrier").release();
+                let started = startup_a.as_mut().await;
+                startup_finished = true;
+                server_a = Some(started.expect("normal A context installation"));
+                assert!(Arc::ptr_eq(&global_b, &resolve_object_store_handle().expect("global remains B")));
+                assert_eq!(connection.peer, server_a.as_ref().expect("A handle").address());
+
+                // A separate source prevents this control from consuming the
+                // delayed request's data and masking an erroneous second lookup.
+                let ready = connection
+                    .rename(signed_rename(&disk_a, USER_VOLUME, "control-source", "ready-control", &control_info))
+                    .await;
+                assert!(ready.success, "a fresh signed user request must actually use Ready: {:?}", ready.error);
+                assert_body(&disk_a, USER_VOLUME, "ready-control", &control_info).await;
+                assert_body(&disk_a, USER_VOLUME, "delayed-source", &delayed_info).await;
+                assert!(!destination_path.exists(), "the original request remains parked");
+                connection.assert_original_connection();
+            }))
+            .catch_unwind()
+            .await;
+
+            // Release on every assertion/timeout path, then drain the original
+            // RPC before shutting down the server and its connection.
+            drop(capture);
+            if let Some(barrier) = startup_barrier.take() {
+                barrier.release();
+            }
+            if !startup_finished {
+                let started = timeout(WAIT, startup_a.as_mut()).await;
+                if let Ok(Ok(started)) = started {
+                    server_a = Some(started);
+                }
+            }
+            let delayed_result = match early_response {
+                Some(response) => Ok(response),
+                None => timeout(WAIT, delayed.as_mut()).await,
+            };
+            let source_after = tokio::fs::read(&source_path).await;
+            let destination_exists = tokio::fs::try_exists(&destination_path).await;
+            let sentinel_after = tokio::fs::read(&sentinel_path).await;
+            (observations, delayed_result, source_after, destination_exists, sentinel_after)
+        };
+        let connection_attempts = connection.attempts.load(Ordering::SeqCst);
+        drop(connection);
+        let shutdown_a = if let Some(server) = server_a {
+            Some(timeout(WAIT, server.shutdown()).await)
+        } else {
+            None
+        };
+        let shutdown_b = timeout(WAIT, server_b.shutdown()).await;
+        if let Some(result) = shutdown_a {
+            result.expect("bounded A shutdown");
+        }
+        shutdown_b.expect("bounded B shutdown");
+
+        assert_eq!(connection_attempts, 1, "the original channel must not redial");
+        match observations {
+            Err(panic) => std::panic::resume_unwind(panic),
+            Ok(result) => result.expect("complete capture/install/Ready-control within the parked request deadline"),
+        }
+        let response = delayed_result
+            .expect("bounded original request drain")
+            .expect("the original signed request must return an application result")
+            .into_inner();
+        assert!(
+            !response.success,
+            "a captured Bootstrap request must not upgrade to Ready after its await"
+        );
+        let error: DiskError = response.error.expect("typed Bootstrap rejection").into();
+        assert_eq!(error, DiskError::FileAccessDenied);
+        assert_eq!(source_after.expect("original source remains readable"), source_before);
+        assert!(!destination_exists.expect("read original destination state"));
+        assert_eq!(sentinel_after.expect("global B sentinel survives"), sentinel_before);
+    }
+}