diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3d12a8b68..bb4a25553 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/crates/e2e_test/src/distributed_startup_regression_test.rs b/crates/e2e_test/src/distributed_startup_regression_test.rs index ff2da755b..3389eb876 100644 --- a/crates/e2e_test/src/distributed_startup_regression_test.rs +++ b/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 { 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>(()) + }) + .await + .map_err(std::io::Error::other)??; + } + Ok::<(), Box>(()) + }) + .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); + + 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 { + 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 { + 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 { + 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> { + 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::(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::>()?; + let events: 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; + } + } } diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 4925b283e..0b9d2871c 100644 --- a/crates/ecstore/Cargo.toml +++ b/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 diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index c4235bd40..3854b2933 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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( pool: Arc, 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( @@ -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 { diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 52a7e6598..151099e8e 100644 --- a/rustfs/Cargo.toml +++ b/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. diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index af8c4e1dd..cd5d8380a 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/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( diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 032e545f1..e8a7ee4fd 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/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 { + 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,