fix(rpc): bind local mutations to the listener instance

.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)
     }
@@ -330,6 +333,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::*;
@@ -1956,7 +2004,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)
     })
@@ -2216,7 +2264,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)
         }
@@ -2349,7 +2397,7 @@ async fn reliable_rename_inner_with_lease(
             let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?;
             #[cfg(all(test, not(windows)))]
             prepared_publication_test_hooks::run_rename_destination(&src_file_path, &dst_file_path);
-            #[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);
+    }
+}