test(rpc): observe bootstrap CAS during fresh startup

.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

crates/e2e_test/src/distributed_startup_regression_test.rs

@@ -195,4 +195,554 @@ mod tests {
         info!("RT-10c PASS: bucket visible from all 4 nodes");
         Ok(())
     }
+
+    /// A real elected CAS must cross a signed peer RPC while the receiver still
+    /// holds its Bootstrap capability, before IAM installs any AppContext.
+    #[tokio::test]
+    async fn test_fresh_four_node_bootstrap_metadata_cas() -> TestResult {
+        use futures::FutureExt;
+        use std::path::PathBuf;
+
+        init_logging();
+        let nonce = uuid::Uuid::new_v4().to_string();
+        let artifact = std::env::var_os("RUSTFS_E2E_STARTUP_CAS_ARTIFACT_DIR")
+            .map(PathBuf::from)
+            .unwrap_or_else(std::env::temp_dir)
+            .join(format!("fresh-startup-cas-{nonce}"));
+        std::fs::create_dir_all(&artifact)?;
+        let binary_dir = tempfile::tempdir()?;
+        let binary = prepare_startup_cas_binary(binary_dir.path(), &artifact)?;
+        probe_startup_cas_binary(&binary, &nonce, &artifact).await?;
+
+        let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
+        let mut logs = Vec::new();
+        let mut disks = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut releases = StartupCasReleases(Vec::new());
+        for i in 0..4 {
+            let disk = PathBuf::from(&cluster.nodes[i].data_dir);
+            assert!(std::fs::read_dir(&disk)?.next().is_none(), "node {i} must start with an empty disk");
+            let log = artifact.join(format!("node-{i}.log"));
+            let release = artifact.join(format!("release-{i}"));
+            cluster.set_node_capture_log_path(i, log.to_string_lossy())?;
+            cluster.set_node_env(i, "RUSTFS_E2E_STARTUP_CAS_RELEASE", release.to_string_lossy())?;
+            endpoints.push(format!("http://{}{}", cluster.nodes[i].address, cluster.nodes[i].data_dir));
+            disks.push(disk);
+            logs.push(log);
+            releases.0.push(release);
+        }
+        assert!(
+            cluster.rustfs_volumes_arg().starts_with(&endpoints[0]),
+            "node 0 must own the elected first endpoint"
+        );
+        cluster.set_env("RUSTFS_E2E_STARTUP_CAS_NONCE", &nonce);
+        cluster.set_env("RUSTFS_OBS_LOG_DIRECTORY", "");
+        cluster.set_env("RUSTFS_OBS_LOG_STDOUT_ENABLED", "true");
+        cluster.set_env("RUST_LOG", "rustfs=info,rustfs_ecstore=trace");
+        for key in [
+            "HTTP_PROXY",
+            "HTTPS_PROXY",
+            "ALL_PROXY",
+            "http_proxy",
+            "https_proxy",
+            "all_proxy",
+        ] {
+            cluster.set_env(key, "");
+        }
+        for key in ["NO_PROXY", "no_proxy"] {
+            cluster.set_env(key, "127.0.0.1,localhost");
+        }
+
+        let attempt = tokio::time::timeout(
+            Duration::from_secs(240),
+            std::panic::AssertUnwindSafe(async {
+                let mut startup = Box::pin(cluster.start_with_binary(&binary));
+                let mut controller = Box::pin(wait_startup_cas(&logs, &disks, &endpoints, &nonce, &artifact));
+                let (observed, startup_finished) = tokio::select! {
+                    observed = tokio::time::timeout(Duration::from_secs(120), &mut controller) => {
+                        (observed.map_err(std::io::Error::other).and_then(|result| result), false)
+                    }
+                    result = &mut startup => {
+                        let error = match result {
+                            Ok(()) => "startup returned before the unreleased CAS gates".to_owned(),
+                            Err(error) => error.to_string(),
+                        };
+                        // Preserve a real pool.bin rejection instead of relabeling
+                        // an earlier identity failure or generic readiness timeout.
+                        let observed = match tokio::time::timeout(Duration::from_secs(5), &mut controller).await {
+                            Ok(Err(error)) => Err(error),
+                            _ => Err(std::io::Error::other(format!("PRECONDITION: startup ended without causal proof: {error}"))),
+                        };
+                        (observed, true)
+                    }
+                };
+                let release_result = releases.release();
+                let drained = if startup_finished {
+                    Ok(())
+                } else {
+                    let deadline = if observed.is_ok() { 60 } else { 5 };
+                    tokio::time::timeout(Duration::from_secs(deadline), &mut startup)
+                        .await
+                        .map_err(std::io::Error::other)
+                        .map_err(|error| -> Box<dyn Error + Send + Sync> { error.into() })
+                        .and_then(|result| result)
+                };
+                drop(startup);
+                observed?;
+                release_result?;
+                drained?;
+                for (i, node) in cluster.nodes.iter().enumerate() {
+                    let pid = node
+                        .process
+                        .as_ref()
+                        .ok_or_else(|| std::io::Error::other("missing child process"))?
+                        .id();
+                    let records = startup_cas_log(&logs[i])?;
+                    assert!(
+                        records
+                            .iter()
+                            .any(|r| r["kind"] == "observer-ready" && r["nonce"] == nonce && r["pid"] == pid),
+                        "node {i} observation must belong to the actual harness child"
+                    );
+                }
+                let bucket = format!("fresh-cas-{nonce}");
+                tokio::time::timeout(Duration::from_secs(10), cluster.create_test_bucket(&bucket))
+                    .await
+                    .map_err(std::io::Error::other)??;
+                for (i, client) in cluster.create_all_clients()?.iter().enumerate() {
+                    let key = format!("node-{i}");
+                    let body = format!("fresh four-node body {i} {nonce}").into_bytes();
+                    tokio::time::timeout(Duration::from_secs(10), async {
+                        client
+                            .put_object()
+                            .bucket(&bucket)
+                            .key(&key)
+                            .body(ByteStream::from(body.clone()))
+                            .send()
+                            .await?;
+                        let received = client
+                            .get_object()
+                            .bucket(&bucket)
+                            .key(&key)
+                            .send()
+                            .await?
+                            .body
+                            .collect()
+                            .await?
+                            .into_bytes();
+                        assert_eq!(received.as_ref(), body, "node {i} must return the complete object body");
+                        Ok::<_, Box<dyn Error + Send + Sync>>(())
+                    })
+                    .await
+                    .map_err(std::io::Error::other)??;
+                }
+                Ok::<(), Box<dyn Error + Send + Sync>>(())
+            })
+            .catch_unwind(),
+        )
+        .await;
+        // Drop the borrowed startup future before stopping its child processes.
+        // All logs are outside the cluster directory which Drop removes.
+        let release_result = releases.release();
+        let pids: Vec<_> = cluster
+            .nodes
+            .iter()
+            .map(|node| node.process.as_ref().map(std::process::Child::id))
+            .collect();
+        let process_record = serde_json::to_vec(&pids)
+            .map_err(std::io::Error::other)
+            .and_then(|bytes| std::fs::write(artifact.join("processes.json"), bytes));
+        cluster.stop();
+        eprintln!("fresh startup CAS evidence: {}", artifact.display());
+        match attempt {
+            Ok(Ok(result)) => {
+                result?;
+                release_result?;
+                process_record?;
+                Ok(())
+            }
+            Ok(Err(panic)) => std::panic::resume_unwind(panic),
+            Err(error) => Err(std::io::Error::other(format!("fresh startup CAS fixture deadline: {error}")).into()),
+        }
+    }
+
+    struct StartupCasReleases(Vec<std::path::PathBuf>);
+
+    impl StartupCasReleases {
+        fn release(&mut self) -> std::io::Result<()> {
+            let mut failure = None;
+            for path in &self.0 {
+                if let Err(error) = std::fs::write(path, b"release") {
+                    failure.get_or_insert(error);
+                }
+            }
+            failure.map_or(Ok(()), Err)
+        }
+    }
+
+    impl Drop for StartupCasReleases {
+        fn drop(&mut self) {
+            let _ = self.release();
+        }
+    }
+
+    fn startup_cas_sha256(path: &std::path::Path) -> std::io::Result<String> {
+        use sha2::{Digest, Sha256};
+        use std::io::Read;
+        let mut file = std::fs::File::open(path)?;
+        let mut hash = Sha256::new();
+        let mut buf = [0; 65536];
+        loop {
+            let len = file.read(&mut buf)?;
+            if len == 0 {
+                break;
+            }
+            hash.update(&buf[..len]);
+        }
+        Ok(format!("{:x}", hash.finalize()))
+    }
+
+    fn startup_cas_git(args: &[&str]) -> std::io::Result<String> {
+        let result = std::process::Command::new("git")
+            .args(args)
+            .current_dir(crate::common::workspace_root())
+            .output()?;
+        if !result.status.success() {
+            return Err(std::io::Error::other("cannot verify startup fixture checkout identity"));
+        }
+        String::from_utf8(result.stdout)
+            .map(|value| value.trim().to_owned())
+            .map_err(std::io::Error::other)
+    }
+
+    fn prepare_startup_cas_binary(dir: &std::path::Path, artifact: &std::path::Path) -> std::io::Result<std::path::PathBuf> {
+        use serde_json::Value;
+        use std::path::PathBuf;
+        let explicit = std::env::var_os("RUSTFS_E2E_STARTUP_CAS_BINARY")
+            .or_else(|| std::env::var_os("CARGO_BIN_EXE_rustfs"))
+            .ok_or_else(|| {
+                std::io::Error::other("PRECONDITION: provide the existing hooks binary; this fixture never invokes Cargo")
+            })?;
+        let binary = std::fs::canonicalize(explicit)?;
+        if let Some(other) = std::env::var_os("CARGO_BIN_EXE_rustfs") {
+            if binary != std::fs::canonicalize(other)? {
+                return Err(std::io::Error::other("PRECONDITION: conflicting startup binary paths"));
+            }
+        }
+        let manifest_path = std::env::var_os("RUSTFS_E2E_STARTUP_CAS_BUILD_MANIFEST")
+            .ok_or_else(|| std::io::Error::other("PRECONDITION: missing hooks binary build manifest"))?;
+        let manifest_bytes = std::fs::read(manifest_path)?;
+        let manifest: Value = serde_json::from_slice(&manifest_bytes)?;
+        std::fs::write(artifact.join("binary-build.json"), &manifest_bytes)?;
+        let checkout = crate::common::workspace_root();
+        let sha = startup_cas_sha256(&binary)?;
+        let valid = manifest["schema"] == 1
+            && manifest["clean_before"] == true
+            && manifest["clean_after"] == true
+            && env!("RUSTFS_E2E_BUILD_DIRTY") == "false"
+            && manifest["commit"] == env!("RUSTFS_E2E_BUILD_COMMIT")
+            && manifest["commit"] == startup_cas_git(&["rev-parse", "HEAD"])?
+            && manifest["tree"] == startup_cas_git(&["rev-parse", "HEAD^{tree}"])?
+            && startup_cas_git(&["status", "--porcelain", "--untracked-files=normal"])?.is_empty()
+            && manifest["lock_git_blob"] == env!("RUSTFS_E2E_BUILD_LOCK")
+            && manifest["lock_git_blob"] == startup_cas_git(&["hash-object", "Cargo.lock"])?
+            && manifest["lock_sha256"] == startup_cas_sha256(&checkout.join("Cargo.lock"))?
+            && manifest["binary_sha256"] == sha
+            && manifest["target"] == env!("RUSTFS_E2E_BUILD_TARGET")
+            && manifest["profile"] == "debug"
+            && manifest["features"]
+                .as_array()
+                .is_some_and(|features| features.iter().any(|f| f == "e2e-test-hooks"))
+            && manifest["argv"]
+                .as_array()
+                .is_some_and(|argv| argv.iter().any(|arg| arg == "--features") && argv.iter().any(|arg| arg == "e2e-test-hooks"))
+            && manifest["rustc_verbose"].as_str().is_some_and(|value| !value.is_empty())
+            && manifest["build_flags"].is_object();
+        if !valid {
+            return Err(std::io::Error::other(
+                "PRECONDITION: hooks binary identity does not match this clean test checkout",
+            ));
+        }
+        let target = dir.join(format!("rustfs{}", std::env::consts::EXE_SUFFIX));
+        std::fs::copy(&binary, &target)?;
+        if startup_cas_sha256(&target)? != sha {
+            return Err(std::io::Error::other("PRECONDITION: binary changed during fixture copy"));
+        }
+        std::fs::write(
+            artifact.join("runner-build.json"),
+            serde_json::to_vec(&serde_json::json!({
+                "commit": env!("RUSTFS_E2E_BUILD_COMMIT"), "lock_git_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
+                "target": env!("RUSTFS_E2E_BUILD_TARGET"), "profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
+                "features": env!("RUSTFS_E2E_BUILD_FEATURES"), "binary_sha256": sha,
+                "binary": PathBuf::from(&target),
+            }))?,
+        )?;
+        Ok(target)
+    }
+
+    async fn probe_startup_cas_binary(binary: &std::path::Path, nonce: &str, artifact: &std::path::Path) -> std::io::Result<()> {
+        struct Probe(std::process::Child);
+        impl Drop for Probe {
+            fn drop(&mut self) {
+                let _ = self.0.kill();
+                let _ = self.0.wait();
+            }
+        }
+        let path = artifact.join("capability-probe.log");
+        let log = std::fs::File::create(&path)?;
+        let mut child = Probe(
+            std::process::Command::new(binary)
+                .arg("--help")
+                .env("RUSTFS_E2E_STARTUP_CAS_PROBE", nonce)
+                .stdout(log.try_clone()?)
+                .stderr(log)
+                .spawn()?,
+        );
+        let status = tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if let Some(status) = child.0.try_wait()? {
+                    return Ok::<_, std::io::Error>(status);
+                }
+                sleep(Duration::from_millis(25)).await;
+            }
+        })
+        .await
+        .map_err(|_| std::io::Error::other("PRECONDITION: binary capability probe timed out"))??;
+        let records = startup_cas_log(&path)?;
+        let matching: Vec<_> = records
+            .iter()
+            .filter(|r| r["nonce"] == nonce && r["kind"] == "capability" && r["schema"] == "fresh-startup-cas/v1")
+            .collect();
+        if !status.success() || matching.len() != 1 {
+            return Err(std::io::Error::other(
+                "PRECONDITION: binary lacks the startup CAS hooks; no cluster was started",
+            ));
+        }
+        Ok(())
+    }
+
+    fn startup_cas_log(path: &std::path::Path) -> std::io::Result<Vec<serde_json::Value>> {
+        let text = match std::fs::read_to_string(path) {
+            Ok(text) => text,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(error) => return Err(error),
+        };
+        if text.len() > 64 * 1024 * 1024 {
+            return Err(std::io::Error::other("startup observation log exceeds fixture bound"));
+        }
+        let mut records = Vec::new();
+        for line in text.split_inclusive('\n').filter_map(|line| line.strip_suffix('\n')) {
+            if let Some(json) = line.strip_prefix("RUSTFS_E2E_STARTUP_CAS ") {
+                records.push(serde_json::from_str(json)?);
+            } else if let Ok(json) = serde_json::from_str::<serde_json::Value>(line) {
+                // The existing remote-disk trace is JSON with flattened fields.
+                records.push(json);
+            }
+        }
+        Ok(records)
+    }
+
+    fn startup_cas_remote_matches(sender: &[serde_json::Value], receiver: &serde_json::Value) -> usize {
+        sender
+            .iter()
+            .filter(|event| {
+                event["target"]
+                    .as_str()
+                    .is_some_and(|target| target.split("::").eq(["rustfs_ecstore", "cluster", "rpc", "remote_disk"]))
+                    && event["op"] == "rename_data"
+                    && event["state"] == "started"
+                    && event["endpoint"] == receiver["disk"]
+                    && ["src_volume", "src_path", "dst_volume", "dst_path"]
+                        .iter()
+                        .all(|key| event[*key] == receiver[*key])
+            })
+            .count()
+    }
+
+    async fn wait_startup_cas(
+        paths: &[std::path::PathBuf],
+        disks: &[std::path::PathBuf],
+        endpoints: &[String],
+        nonce: &str,
+        artifact: &std::path::Path,
+    ) -> std::io::Result<()> {
+        loop {
+            let logs: Vec<_> = paths
+                .iter()
+                .map(|path| startup_cas_log(path))
+                .collect::<std::io::Result<_>>()?;
+            let events: Vec<Vec<_>> = logs
+                .iter()
+                .map(|records| records.iter().filter(|r| r["nonce"] == nonce).collect())
+                .collect();
+            let source = &events[0];
+            for (node, events) in events.iter().enumerate() {
+                for event in events.iter().filter(|r| r["kind"] == "cas") {
+                    if node != 0 {
+                        return Err(std::io::Error::other(format!(
+                            "PRECONDITION: non-elected node {node} executed CAS: {event}"
+                        )));
+                    }
+                    if event["ok"] == false {
+                        let object = event["object"].as_str().unwrap_or_default();
+                        let receiver = logs.iter().skip(1).flat_map(|records| records.iter()).find(|r| {
+                            r["nonce"] == nonce
+                                && r["kind"] == "receiver"
+                                && r["dst_path"] == object
+                                && r["ok"] == false
+                                && r["target"] == "bootstrap"
+                                && startup_cas_remote_matches(&logs[0], r) == 1
+                        });
+                        if let Some(receiver) = receiver {
+                            let identity_ok = source
+                                .iter()
+                                .take_while(|r| !std::ptr::eq(**r, *event))
+                                .any(|r| r["kind"] == "cas" && r["phase"] == "identity_cas" && r["ok"] == true);
+                            let class = if object == "pool.bin" && identity_ok {
+                                "POOL_BIN_CAUSAL_REJECTION"
+                            } else {
+                                "PRECONDITION_IDENTITY_OR_STARTUP_FAILURE"
+                            };
+                            std::fs::write(
+                                artifact.join("cas-rejection.json"),
+                                serde_json::to_vec_pretty(&serde_json::json!({
+                                    "class": class, "sender": event, "receiver": receiver,
+                                }))?,
+                            )?;
+                            return Err(std::io::Error::other(format!("{class}: sender={event}; receiver={receiver}")));
+                        }
+                    }
+                }
+            }
+            if events
+                .iter()
+                .all(|records| records.iter().any(|r| r["kind"] == "gate" && r["slot_installed"] == false))
+            {
+                let mut pids = std::collections::HashSet::new();
+                for records in &events {
+                    let ready: Vec<_> = records.iter().filter(|r| r["kind"] == "observer-ready").collect();
+                    if ready.len() != 1
+                        || !pids.insert(
+                            ready[0]["pid"]
+                                .as_u64()
+                                .ok_or_else(|| std::io::Error::other("missing child PID"))?,
+                        )
+                    {
+                        return Err(std::io::Error::other(
+                            "PRECONDITION: four independent observer-capable child processes required",
+                        ));
+                    }
+                    if records.iter().any(|r| r["pid"] != ready[0]["pid"]) {
+                        return Err(std::io::Error::other("PRECONDITION: observation process mismatch"));
+                    }
+                }
+                let prepare: Vec<_> = source
+                    .iter()
+                    .filter(|r| r["kind"] == "cas" && r["phase"] == "prepare_cas" && r["object"] == "pool.bin")
+                    .collect();
+                let commit: Vec<_> = source
+                    .iter()
+                    .filter(|r| r["kind"] == "cas" && r["phase"] == "commit_cas" && r["object"] == "pool.bin")
+                    .collect();
+                if prepare.len() != 1 || commit.len() != 1 {
+                    return Err(std::io::Error::other("PRECONDITION: startup CAS evidence is absent or ambiguous"));
+                }
+                let (prepare, commit) = (*prepare[0], *commit[0]);
+                for cas in [prepare, commit] {
+                    if cas["ok"] != true
+                        || cas["tail_drained"] != true
+                        || cas["no_lock"] != true
+                        || cas["etag"].as_str().is_none_or(str::is_empty)
+                    {
+                        return Err(std::io::Error::other(format!("actual startup CAS did not complete: {cas}")));
+                    }
+                }
+                if prepare["if_none_match"] != "*"
+                    || !prepare["if_match"].is_null()
+                    || commit["if_match"] != prepare["etag"]
+                    || !commit["if_none_match"].is_null()
+                    || prepare["etag"] == commit["etag"]
+                    || prepare["payload_sha256"] == commit["payload_sha256"]
+                {
+                    return Err(std::io::Error::other(
+                        "actual prepare/commit conditional revisions do not form the fresh CAS chain",
+                    ));
+                }
+                let confirmed = source.iter().find(|r| {
+                    r["kind"] == "confirmed"
+                        && r["payload_sha256"] == commit["payload_sha256"]
+                        && r["generation"].as_u64().is_some_and(|g| g > 0)
+                });
+                if confirmed.is_none() {
+                    return Err(std::io::Error::other("actual quorum reload did not confirm the committed payload"));
+                }
+                for node in 1..4 {
+                    let mut accepted = Vec::new();
+                    for cas in [prepare, commit] {
+                        let matching: Vec<_> = events[node]
+                            .iter()
+                            .filter(|r| {
+                                r["kind"] == "receiver"
+                                    && r["dst_volume"] == ".rustfs.sys"
+                                    && r["dst_path"] == "pool.bin"
+                                    && r["etag"] == cas["etag"]
+                            })
+                            .collect();
+                        if matching.len() != 1 {
+                            break;
+                        }
+                        let received = *matching[0];
+                        if received["ok"] != true
+                            || received["target"] != "bootstrap"
+                            || received["disk"] != endpoints[node]
+                            || received["body_sha256"].as_str().is_none_or(|hash| hash.len() != 64)
+                            || startup_cas_remote_matches(&logs[0], received) != 1
+                        {
+                            break;
+                        }
+                        accepted.push(received);
+                    }
+                    if accepted.len() == 2 {
+                        let raw = std::fs::read(disks[node].join(".rustfs.sys/pool.bin/xl.meta"))?;
+                        std::fs::write(artifact.join(format!("node-{node}-committed-xl.meta")), &raw)?;
+                        let file_info = rustfs_filemeta::get_file_info(
+                            &raw,
+                            ".rustfs.sys",
+                            "pool.bin",
+                            "",
+                            rustfs_filemeta::FileInfoOpts {
+                                data: false,
+                                include_free_versions: false,
+                                include_part_checksums: false,
+                            },
+                        )
+                        .map_err(std::io::Error::other)?;
+                        if raw.is_empty()
+                            || file_info.metadata.get("etag").map(String::as_str) != commit["etag"].as_str()
+                            || file_info
+                                .mod_time
+                                .map(|time| time.unix_timestamp_nanos().to_string())
+                                .as_deref()
+                                != accepted[1]["mod_time"].as_str()
+                            || accepted[1]["mod_time"].is_null()
+                        {
+                            return Err(std::io::Error::other(
+                                "latest physical target metadata does not match the accepted commit",
+                            ));
+                        }
+                        std::fs::write(
+                            artifact.join("cas-proof.json"),
+                            serde_json::to_vec_pretty(&serde_json::json!({
+                                "sender": 0, "receiver": node, "prepare": prepare, "commit": commit, "accepted": accepted, "confirmed": confirmed,
+                            }))?,
+                        )?;
+                        return Ok(());
+                    }
+                }
+                // A started trace may still be flushing asynchronously. Keep
+                // waiting for its real tuple; the enclosing deadline is finite.
+            }
+            sleep(Duration::from_millis(25)).await;
+        }
+    }
 }

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/core/pools.rs

@@ -5480,6 +5480,21 @@ fn pool_meta_cas_preconditions(token: &PoolMetaCasToken, object: &str) -> Result
     }
 }
 
+// Direct JSON diagnostics are independent of the startup tracing subscriber.
+#[cfg(feature = "e2e-test-hooks")]
+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 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 +5515,33 @@ 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,
+            "payload_sha256": format!("{:x}", 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["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 +6840,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": format!("{:x}", Sha256::digest(&durable)),
+                "generation": confirmed.revision.generation,
+                "transaction_id": confirmed.revision.transaction_id,
+            }));
             return Ok(confirmed.meta);
         }
         if !commit_succeeded {

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/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
@@ -160,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/disk.rs

@@ -39,6 +39,29 @@ 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": format!("{:x}", 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,
@@ -1275,7 +1298,9 @@ impl NodeService {
             Some(token)
         };
         let request_decoded_from_msgpack = decoded_file_info.from_msgpack;
-        match target
+        #[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),
@@ -1283,8 +1308,15 @@ impl NodeService {
                 (&request.dst_volume, &request.dst_path),
                 scanner_publication_lease_token,
             )
-            .await
-        {
+            .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,