Compare commits

...

23 Commits

Author SHA1 Message Date
overtrue 6233466c5d chore: integrate current main for namespace target validation 2026-09-06 15:49:36 +08:00
overtrue 901d052c38 test(e2e): declare tempfile for fresh startup fixture 2026-09-06 15:45:26 +08:00
overtrue fdb3d9f9bb test(startup): encode observation digests with shared hex helper 2026-09-06 15:10:28 +08:00
overtrue 44333e136b test(rpc): use native part size in signed fixture 2026-09-06 13:57:07 +08:00
overtrue 9e24d23c30 test(rpc): observe bootstrap CAS during fresh startup 2026-09-06 13:01:22 +08:00
overtrue 6277287399 chore(test): merge main before startup CAS coverage 2026-09-06 12:45:50 +08:00
overtrue 358ff0f0e6 test(rpc): select fixture disks through the store topology 2026-09-06 11:42:53 +08:00
overtrue 1cea5fa1c4 test(rpc): retain bootstrap authority across delayed requests 2026-09-06 11:42:53 +08:00
overtrue b1a2235cd6 fix(ecstore): keep group fsync hooks scoped to unit tests 2026-09-06 11:25:31 +08:00
overtrue e0663c11af test(rpc): cover signed listener binding across startup 2026-09-06 11:13:09 +08:00
overtrue 11c9fd64ce fix(rpc): bind local mutations to the listener instance 2026-09-06 10:58:11 +08:00
overtrue 7f631ec378 test(rpc): mark instance regression request as v2 authenticated 2026-09-06 10:21:55 +08:00
overtrue 3205f85c2a test(rpc): preserve bootstrap metadata in instance snapshot 2026-09-06 10:21:55 +08:00
overtrue ae87ddbe2f test(rpc): reproduce same-UUID cross-instance rename 2026-09-06 10:21:55 +08:00
overtrue f09aaad2a9 Merge local namespace ownership prerequisite 2026-09-06 10:21:54 +08:00
overtrue 3ae29aab26 test(ecstore): wait for namespace owner release before asserting
The namespace owner tests decided that ownership had ended when the Weak probe stopped upgrading or when the mutation lease could be reacquired. Both signals fire before the owner guard's Drop decrements the pending counter: Arc releases its strong count before running Drop, and the lease drops its locks before its owner field. The rio-v2 lane hit that window in undo_fresh_version_keeps_physical_namespace_owner_after_timeout.

Extend every drain wait to also require namespace_commits_pending() to be false, so the assertions observe the completed release instead of racing it.
2026-09-06 10:09:44 +08:00
overtrue 3c54ac1deb Merge remote-tracking branch 'origin/main' into overtrue/fix/fsync-group-identity-capture 2026-09-06 04:49:00 +08:00
overtrue 44fe0b9950 Merge remote-tracking branch 'origin/main' into overtrue/fix/fsync-group-identity-capture 2026-09-06 04:44:25 +08:00
overtrue 58f1840630 test(ecstore): mark physical owner fixtures as inline 2026-09-06 04:25:12 +08:00
overtrue 5e33186cf6 fix(ecstore): preserve successor fsync group registration
(cherry picked from commit c7dfaad90526052e56c57dafffa4813bdcde46ca)
2026-09-06 03:54:07 +08:00
overtrue ed100103d0 fix(ecstore): capture complete fsync worker guard
(cherry picked from commit 1dc90bb836e20ea9ee45d0a629a9201e20d231c4)
2026-09-06 03:54:07 +08:00
overtrue 93c89ef132 test(ecstore): expose stale fsync group cleanup 2026-09-06 03:20:33 +08:00
overtrue ea4068b8ac fix(ecstore): retain namespace owners through local physical tails
(cherry picked from commit a2f242463316e87604feadbdac5e4148140e72c0)
2026-09-06 03:17:00 +08:00
23 changed files with 3080 additions and 162 deletions
+93 -2
View File
@@ -582,13 +582,59 @@ jobs:
install-build-packaging-tools: 'false' install-build-packaging-tools: 'false'
- name: Build debug binary - 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 - name: Upload debug binary
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6 uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
with: with:
name: rustfs-debug-binary 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 if-no-files-found: error
retention-days: 1 retention-days: 1
@@ -906,6 +952,36 @@ jobs:
- name: Make binary executable - name: Make binary executable
run: chmod +x ./target/debug/rustfs 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 - name: Verify e2e full membership
env: env:
NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-full-list.json 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 # 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. # debug binary; each test spawns its own rustfs server on a random port.
- name: Run e2e full suite - 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 run: cargo nextest run --profile e2e-full -p e2e_test
- name: Upload junit - name: Upload junit
@@ -930,6 +1010,17 @@ jobs:
${{ runner.temp }}/rustfs-e2e-full-list.json ${{ runner.temp }}/rustfs-e2e-full-list.json
retention-days: 7 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: e2e-tests-rio-v2:
name: End-to-End Tests (rio-v2) name: End-to-End Tests (rio-v2)
# Inherits the schedule/dispatch-only gate through needs: on every other # Inherits the schedule/dispatch-only gate through needs: on every other
Generated
+1
View File
@@ -4057,6 +4057,7 @@ dependencies = [
"sha1 0.11.0", "sha1 0.11.0",
"sha2 0.11.0", "sha2 0.11.0",
"suppaftp", "suppaftp",
"tempfile",
"time", "time",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
+3
View File
@@ -144,3 +144,6 @@ russh = { workspace = true, features = ["serde"] }
russh-sftp = { workspace = true } russh-sftp = { workspace = true }
zip.workspace = true zip.workspace = true
clap = { workspace = true, features = ["derive", "env"] } clap = { workspace = true, features = ["derive", "env"] }
[dev-dependencies]
tempfile.workspace = true
@@ -195,4 +195,554 @@ mod tests {
info!("RT-10c PASS: bucket visible from all 4 nodes"); info!("RT-10c PASS: bucket visible from all 4 nodes");
Ok(()) 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(rustfs_utils::crypto::hex(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;
}
}
} }
+2
View File
@@ -118,6 +118,8 @@ hotpath-cpu = [
# injection, xl.meta transition assertions) via `api::tier::test_util`. # injection, xl.meta transition assertions) via `api::tier::test_util`.
# Enable only from `[dev-dependencies]` (rustfs/backlog#1148 ilm-6). # Enable only from `[dev-dependencies]` (rustfs/backlog#1148 ilm-6).
test-util = [] test-util = []
# Observes real startup CAS only in the dedicated E2E binary.
e2e-test-hooks = []
[dependencies] [dependencies]
hotpath.workspace = true hotpath.workspace = true
+4 -2
View File
@@ -368,6 +368,8 @@ pub mod data_usage {
pub mod disk { pub mod disk {
pub use crate::disk::disk_store::get_object_disk_read_timeout; pub use crate::disk::disk_store::get_object_disk_read_timeout;
pub use crate::disk::local::ScanGuard; pub use crate::disk::local::ScanGuard;
#[cfg(all(feature = "test-util", not(windows)))]
pub use crate::disk::os::{LocalPublicationPause, LocalPublicationStage};
pub use crate::disk::{ pub use crate::disk::{
BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp, BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
CheckPartsResp, ConditionalFileUpdate, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, 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::core::pools::HealLifecycleExpiryContext;
pub use crate::store::HealWalkVersion; pub use crate::store::HealWalkVersion;
pub use crate::store::{ pub use crate::store::{
ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, all_local_disk_path, BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
find_local_disk_by_ref, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients, 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, prewarm_local_disk_id_map, prewarm_local_disk_id_map_with_instance_ctx,
}; };
} }
+45 -3
View File
@@ -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>( async fn save_pool_meta_object_cas<S>(
pool: Arc<S>, pool: Arc<S>,
object: &str, object: &str,
@@ -5500,13 +5515,33 @@ where
..Default::default() ..Default::default()
}; };
fence.add_to_options(&mut opts); 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": 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; let result = save_config_with_opts_and_metadata(pool, object, data, &opts).await;
if matches!(&result, Err(Error::PreconditionFailed)) { if matches!(&result, Err(Error::PreconditionFailed)) {
record_pool_meta_stale_write_rejection(phase); record_pool_meta_stale_write_rejection(phase);
} }
let object_info = result?; let result = result.and_then(|object_info| {
fence.ensure_held()?; fence.ensure_held()?;
Ok(object_info) 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>( async fn persist_pool_meta_identity<S>(
@@ -6805,6 +6840,13 @@ impl PoolMeta {
}; };
if confirmed.revision == revision && confirmed.canonical.as_ref() == Some(&durable) { if confirmed.revision == revision && confirmed.canonical.as_ref() == Some(&durable) {
persist_pool_meta_identity(pools, write_state, true, fence).await?; 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); return Ok(confirmed.meta);
} }
if !commit_succeeded { if !commit_succeeded {
+52 -4
View File
@@ -247,7 +247,7 @@ pub(crate) mod fsync_dir_recorder {
} }
/// Pause a real namespace mutation inside its physical executor. /// 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 { pub(crate) mod prepared_publication_test_hooks {
use super::*; use super::*;
@@ -256,7 +256,9 @@ pub(crate) mod prepared_publication_test_hooks {
PreparedRename, PreparedRename,
Rename, Rename,
Remove, Remove,
#[cfg(test)]
Rollback, Rollback,
#[cfg(test)]
DirFsync, 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 { pub(crate) fn install(path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard {
install_at(Stage::PreparedRename, path, hook) install_at(Stage::PreparedRename, path, hook)
} }
@@ -290,6 +293,51 @@ pub(crate) mod prepared_publication_test_hooks {
} }
} }
/// Controlled application-test pause at an existing physical executor boundary.
#[cfg(all(feature = "test-util", not(windows)))]
pub struct LocalPublicationPause {
_hook: prepared_publication_test_hooks::Guard,
entered: oneshot::Receiver<()>,
_release: std::sync::mpsc::Sender<()>,
}
#[cfg(all(feature = "test-util", not(windows)))]
#[derive(Clone, Copy)]
pub enum LocalPublicationStage {
PreparedRename,
Rename,
Remove,
}
#[cfg(all(feature = "test-util", not(windows)))]
impl LocalPublicationPause {
pub fn install(disk: &crate::disk::Disk, volume: &str, path: &str, stage: LocalPublicationStage) -> Result<Self> {
let path = disk
.get_object_path_for_io_if_local(volume, path)
.ok_or(DiskError::DiskNotFound)??;
let stage = match stage {
LocalPublicationStage::PreparedRename => prepared_publication_test_hooks::Stage::PreparedRename,
LocalPublicationStage::Rename => prepared_publication_test_hooks::Stage::Rename,
LocalPublicationStage::Remove => prepared_publication_test_hooks::Stage::Remove,
};
let (entered_tx, entered) = oneshot::channel();
let (release, release_rx) = std::sync::mpsc::channel::<()>();
let hook = prepared_publication_test_hooks::install_at(stage, &path, move || {
let _ = entered_tx.send(());
let _ = release_rx.recv();
});
Ok(Self {
_hook: hook,
entered,
_release: release,
})
}
pub async fn entered(&mut self) -> std::result::Result<(), oneshot::error::RecvError> {
(&mut self.entered).await
}
}
#[cfg(all(test, windows))] #[cfg(all(test, windows))]
pub(crate) mod windows_rename_test_hooks { pub(crate) mod windows_rename_test_hooks {
use super::*; use super::*;
@@ -1916,7 +1964,7 @@ pub(crate) async fn remove_file_with_owner(
let path = path.as_ref().to_path_buf(); let path = path.as_ref().to_path_buf();
let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await; let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
run_blocking_namespace_operation(lease, move || { 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); prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Remove, &path);
std::fs::remove_file(path) std::fs::remove_file(path)
}) })
@@ -2140,7 +2188,7 @@ pub(crate) async fn rename_all_with_prepared_source(
move || { move || {
validate_prepared_rename_source(&prepared_source, &src_file_path)?; 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)?; 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); prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path);
rename_prepared(&src_file_path, &dst_file_path, &preparation) rename_prepared(&src_file_path, &dst_file_path, &preparation)
} }
@@ -2271,7 +2319,7 @@ async fn reliable_rename_inner_with_lease(
let base_dir = base_dir.clone(); let base_dir = base_dir.clone();
move || { move || {
let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?; let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?;
#[cfg(all(test, not(windows)))] #[cfg(all(any(test, feature = "test-util"), not(windows)))]
{ {
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src_file_path); prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src_file_path);
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path); prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path);
+2 -2
View File
@@ -442,7 +442,7 @@ pub(crate) mod utils;
use peer::init_local_peer; use peer::init_local_peer;
pub use 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, init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
prewarm_local_disk_id_map_with_instance_ctx, prewarm_local_disk_id_map_with_instance_ctx,
}; };
@@ -1787,7 +1787,7 @@ mod tests {
// Build a minimal ECStore carrying an explicit instance context. Empty // Build a minimal ECStore carrying an explicit instance context. Empty
// pools/disks are sufficient: the Phase 5 accessors read only `self.ctx`. // 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(); let endpoint_pools = EndpointServerPools::default();
Arc::new(ECStore { Arc::new(ECStore {
id: uuid::Uuid::new_v4(), id: uuid::Uuid::new_v4(),
+717 -1
View File
@@ -13,7 +13,10 @@
// limitations under the License. // limitations under the License.
use super::*; 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 crate::runtime::sources as runtime_sources;
use tracing::{debug, error}; 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_LOCAL_DISK_ID_PREWARM_SKIPPED: &str = "local_disk_id_prewarm_skipped";
const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initialization_failed"; 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> { async fn remember_local_disk_id(disk: &DiskStore) -> Option<Uuid> {
remember_local_disk_id_with_instance_ctx(&crate::runtime::global::current_ctx(), disk).await 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] #[test]
fn endpoint_rpc_authority_preserves_port_and_ipv6_brackets() { fn endpoint_rpc_authority_preserves_port_and_ipv6_brackets() {
let endpoint = Endpoint::try_from("https://127.0.0.1:9001/d1").expect("URL endpoint"); let endpoint = Endpoint::try_from("https://127.0.0.1:9001/d1").expect("URL endpoint");
+1 -1
View File
@@ -68,7 +68,7 @@ license = []
io-scheduler-debug = [] # Enable debug information in I/O scheduler 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) tracing-chunk-debug = [] # Enable per-chunk tracing in data plane (high noise, for debugging only)
full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp", "pyroscope", "gcs"] 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. # Shortens Connect credentials only in debug E2E builds.
connect-e2e-short-credentials = [] connect-e2e-short-credentials = []
# Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root. # Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root.
+35 -2
View File
@@ -26,13 +26,14 @@
//! server is not ready rather than that another server's global context applies. //! server is not ready rather than that another server's global context applies.
use super::global::{AppContext, get_global_app_context}; 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}; use std::sync::{Arc, OnceLock};
/// Late-bound, per-server handle to the application context. /// Late-bound, per-server handle to the application context.
#[derive(Default)] #[derive(Default)]
pub struct ServerContextSlot { pub struct ServerContextSlot {
app_context: OnceLock<Arc<AppContext>>, app_context: OnceLock<Arc<AppContext>>,
bootstrap_target: Option<BootstrapLocalTarget>,
heal_topology_fingerprint: Arc<tokio::sync::OnceCell<String>>, heal_topology_fingerprint: Arc<tokio::sync::OnceCell<String>>,
} }
@@ -50,15 +51,47 @@ impl ServerContextSlot {
pub fn new() -> Arc<Self> { pub fn new() -> Arc<Self> {
Arc::new(Self { Arc::new(Self {
app_context: OnceLock::new(), app_context: OnceLock::new(),
bootstrap_target: None,
heal_topology_fingerprint: Arc::new(tokio::sync::OnceCell::new()), 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 /// Install this server's application context (once). Returns `false` if
/// the slot was already installed; the first installation wins, matching /// the slot was already installed; the first installation wins, matching
/// the process-global singleton's `get_or_init` semantics. /// the process-global singleton's `get_or_init` semantics.
pub fn install(&self, context: Arc<AppContext>) -> bool { 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. /// This server's installed application context, if startup has completed.
+2 -2
View File
@@ -37,8 +37,8 @@ impl AppContext {
// also publishes to the process default (first server wins) so legacy // also publishes to the process default (first server wins) so legacy
// free-function readers keep resolving the first server's context. // free-function readers keep resolving the first server's context.
let context = Arc::new(AppContext::with_default_interfaces(store, iam, kms_interface)); let context = Arc::new(AppContext::with_default_interfaces(store, iam, kms_interface));
publish_global_app_context(context.clone()); server_ctx.try_install(context.clone())?;
let _ = server_ctx.install(context); publish_global_app_context(context);
Ok(()) Ok(())
} }
} }
+1 -1
View File
@@ -1261,7 +1261,7 @@ pub(crate) mod context {
pub(crate) use super::EndpointServerPools; pub(crate) use super::EndpointServerPools;
pub(crate) use super::bucket; pub(crate) use super::bucket;
pub(crate) use super::runtime; 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)] #[cfg(test)]
pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints}; pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints};
} }
+3 -1
View File
@@ -36,7 +36,9 @@ use crate::server::{
}; };
use crate::storage_api::server::http as storage; use crate::storage_api::server::http as storage;
use crate::storage_api::server::http::rpc::InternodeRpcService; 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;
use crate::storage_api::server::http::tonic_service::make_server_for_slot;
use crate::storage_api::server::http::{ use crate::storage_api::server::http::{
ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge, ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge,
tonic_boot_epoch_response_headers, verify_tonic_rpc_signature_with_bootstrap, tonic_boot_epoch_response_headers, verify_tonic_rpc_signature_with_bootstrap,
@@ -1834,7 +1836,7 @@ fn process_connection(
// each service in the auth interceptor. // each service in the auth interceptor.
let rpc_max_message_size = rustfs_protos::internode_rpc_max_message_size(); let rpc_max_message_size = rustfs_protos::internode_rpc_max_message_size();
let node_service = InterceptedService::new( 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_decoding_message_size(rpc_max_message_size)
.max_encoding_message_size(rpc_max_message_size), .max_encoding_message_size(rpc_max_message_size),
check_auth, check_auth,
+1 -3
View File
@@ -124,9 +124,6 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
} else { } else {
bootstrap_instance_ctx() bootstrap_instance_ctx()
}; };
// This server's request-path context slot (backlog#1052 S2).
let server_ctx = ServerContextSlot::new();
let EmbeddedStartupConfig { let EmbeddedStartupConfig {
config, config,
identity, identity,
@@ -151,6 +148,7 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
.await .await
.map_err(init_error)?; .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 http_server = start_embedded_http_server(&config, listen_context.readiness.clone(), server_ctx.clone()).await?;
let shutdown_handle = http_server.shutdown_handle; let shutdown_handle = http_server.shutdown_handle;
let bound_addr = http_server.bound_addr; let bound_addr = http_server.bound_addr;
+51 -4
View File
@@ -62,6 +62,29 @@ fn emit_fatal_stderr(context: &str, error: impl std::fmt::Display) {
} }
async fn async_main() -> Result<()> { 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!(); hotpath::tokio_runtime!();
// Log container resource detection early in startup // 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 // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future
// multi-instance server constructs its own context here instead. // multi-instance server constructs its own context here instead.
let instance_ctx = bootstrap_instance_ctx(); 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 { let StartupListenContext {
readiness, readiness,
server_addr, server_addr,
@@ -152,6 +171,7 @@ async fn run(config: Config) -> Result<()> {
} = init_startup_listen_context(&config, &instance_ctx).await?; } = init_startup_listen_context(&config, &instance_ctx).await?;
let endpoint_pools = init_startup_storage_foundation(&server_address, &config.volumes, &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 { let StartupHttpServers {
state_manager, state_manager,
s3_shutdown_tx, s3_shutdown_tx,
@@ -163,6 +183,33 @@ async fn run(config: Config) -> Result<()> {
shutdown_token: ctx, shutdown_token: ctx,
} = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone(), instance_ctx).await?; } = 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 capacity_tasks = crate::capacity::capacity_integration::init_capacity_management_managed().await;
let service_runtime = init_startup_runtime_services( let service_runtime = init_startup_runtime_services(
+806 -3
View File
@@ -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 _, 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, 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::{ 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, verify_tonic_mutation_body_digest_reject_unsigned,
}; };
use bytes::Bytes; use bytes::Bytes;
@@ -482,6 +482,97 @@ mod metrics;
pub struct NodeService { pub struct NodeService {
local_peer: LocalPeerS3Client, local_peer: LocalPeerS3Client,
context: Option<Arc<runtime_sources::AppContext>>, 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 { 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 { pub fn make_server_for_context(context: Option<Arc<runtime_sources::AppContext>>) -> NodeService {
let local_peer = LocalPeerS3Client::new(None, None); 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)] #[derive(Clone, Debug, Default)]
@@ -1074,6 +1177,24 @@ impl heal_control_service_server::HealControlService for HealControlRpcService {
} }
impl NodeService { 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>> { fn resolve_object_store(&self) -> Option<Arc<ECStore>> {
let context = self.context.clone().or_else(runtime_sources::current_app_context); let context = self.context.clone().or_else(runtime_sources::current_app_context);
runtime_sources::current_object_store_handle_for_context(context.as_deref()) runtime_sources::current_object_store_handle_for_context(context.as_deref())
@@ -2680,6 +2801,7 @@ mod tests {
validate_admin_heal_control_start, validate_admin_heal_control_start,
}; };
use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; 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::rpc_consumer::node_service::{DiskError, HealBucketInfo};
use crate::storage::storage_api::set_tonic_canonical_body_digest; use crate::storage::storage_api::set_tonic_canonical_body_digest;
use crate::storage::storage_api::{ use crate::storage::storage_api::{
@@ -4660,6 +4782,687 @@ mod tests {
assert!(rename_response.error.is_some()); 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] #[tokio::test]
async fn test_make_volumes_invalid_disk() { async fn test_make_volumes_invalid_disk() {
let service = create_test_node_service(); let service = create_test_node_service();
+165 -126
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and // See the License for the specific language governing permissions and
// limitations under the License. // limitations under the License.
use super::NodeService; use super::{LocalMutationTarget, NodeService};
use crate::storage::storage_api::rpc_consumer::node_service::{ use crate::storage::storage_api::rpc_consumer::node_service::{
BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq, BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq,
ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count, ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count,
@@ -39,6 +39,69 @@ use tonic::{Request, Response, Status};
use tracing::debug; use tracing::debug;
use uuid::Uuid; 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. /// Initial capacity hint (bytes) for typical small msgpack requests and responses.
const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512; const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024; const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024;
@@ -670,55 +733,59 @@ impl NodeService {
"delete_version", "delete_version",
)?; )?;
let request = request.into_inner(); 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") {
let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") { Ok(file_info) => file_info,
Ok(file_info) => file_info, Err(err) => {
Err(err) => { return Ok(Response::new(DeleteVersionResponse {
return Ok(Response::new(DeleteVersionResponse { success: false,
success: false, raw_file_info: "".to_string(),
raw_file_info: "".to_string(), error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()),
error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()), }));
})); }
} };
}; let opts = match decode_msgpack_or_json::<DeleteOptions>(&request.opts_bin, &request.opts, "DeleteOptions") {
let opts = match decode_msgpack_or_json::<DeleteOptions>(&request.opts_bin, &request.opts, "DeleteOptions") { Ok(opts) => opts,
Ok(opts) => opts, Err(err) => {
Err(err) => { return Ok(Response::new(DeleteVersionResponse {
return Ok(Response::new(DeleteVersionResponse { success: false,
success: false, raw_file_info: "".to_string(),
raw_file_info: "".to_string(), error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()), }));
})); }
} };
}; let result = if opts.undo_write {
match disk if request.force_del_marker {
.delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts) 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 .await
{ } else {
Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) { Err(DiskError::other("cannot find disk"))
Ok(raw_file_info) => Ok(Response::new(DeleteVersionResponse { };
success: true, match result {
raw_file_info, Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) {
error: None, Ok(raw_file_info) => Ok(Response::new(DeleteVersionResponse {
})), success: true,
Err(err) => Ok(Response::new(DeleteVersionResponse { raw_file_info,
success: false, error: None,
raw_file_info: "".to_string(), })),
error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
})),
},
Err(err) => Ok(Response::new(DeleteVersionResponse { Err(err) => Ok(Response::new(DeleteVersionResponse {
success: false, success: false,
raw_file_info: "".to_string(), raw_file_info: "".to_string(),
error: Some(err.into()), error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
})), })),
} },
} else { Err(err) => Ok(Response::new(DeleteVersionResponse {
Ok(Response::new(DeleteVersionResponse {
success: false, success: false,
raw_file_info: "".to_string(), 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", "rename_data",
)?; )?;
let request = request.into_inner(); let request = request.into_inner();
if let Some(disk) = self.find_disk(&request.disk).await { let target = self.local_mutation_target();
let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) { #[cfg(feature = "e2e-test-hooks")]
Ok(file_info) => file_info, super::rename_target_capture_test_hook::wait(&target, &request).await;
Err(err) => { let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) {
return Ok(Response::new(RenameDataResponse { Ok(file_info) => file_info,
success: false, Err(err) => {
rename_data_resp: String::new(), return Ok(Response::new(RenameDataResponse {
rename_data_resp_bin: Vec::new().into(), success: false,
error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()), 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 scanner_publication_lease_token = if request.scanner_publication_lease_token.is_empty() {
let token = Uuid::from_slice(&request.scanner_publication_lease_token) None
.map_err(|_| Status::invalid_argument("scanner publication lease token must be a UUID"))?; } else {
if token.is_nil() { let token = Uuid::from_slice(&request.scanner_publication_lease_token)
return Err(Status::invalid_argument("scanner publication lease token must not be nil")); .map_err(|_| Status::invalid_argument("scanner publication lease token must be a UUID"))?;
} if token.is_nil() {
Some(token) return Err(Status::invalid_argument("scanner publication lease token must not be nil"));
}; }
// The target owns this read guard. It must span the complete Some(token)
// disk rename, not merely the preflight, so a movement transition };
// cannot restart after validation and before rename linearization. let request_decoded_from_msgpack = decoded_file_info.from_msgpack;
let scanner_publication_lease_guard: Option<Arc<dyn Send + Sync>> = #[cfg(feature = "e2e-test-hooks")]
if let Some(token) = scanner_publication_lease_token { let observation = startup_cas_rename_observation(&target, &request, &decoded_file_info.value);
let Some(store) = self.resolve_object_store() else { let result = target
return Ok(Response::new(RenameDataResponse { .rename_local_data(
success: false, &request.disk,
rename_data_resp: String::new(), (&request.src_volume, &request.src_path),
rename_data_resp_bin: Vec::new().into(), &decoded_file_info.value,
error: Some(DiskError::other("scanner publication lease owner is unavailable").into()), (&request.dst_volume, &request.dst_path),
})); scanner_publication_lease_token,
}; )
match store.acquire_scanner_publication_lease_guard(token).await { .await;
Ok(guard) => Some(Arc::new(guard)), #[cfg(feature = "e2e-test-hooks")]
Err(err) => { if let Some(mut observation) = observation {
return Ok(Response::new(RenameDataResponse { observation["ok"] = serde_json::json!(result.is_ok());
success: false, observation["error"] = serde_json::json!(result.as_ref().err().map(ToString::to_string));
rename_data_resp: String::new(), let line = format!("RUSTFS_E2E_STARTUP_CAS {observation}\n");
rename_data_resp_bin: Vec::new().into(), let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes());
error: Some(DiskError::other(err.to_string()).into()), }
})); 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 {
} else { success: true,
None rename_data_resp,
}; rename_data_resp_bin: rename_data_resp_bin.into(),
let request_decoded_from_msgpack = decoded_file_info.from_msgpack; error: None,
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()),
})),
}
}
Err(err) => Ok(Response::new(RenameDataResponse { Err(err) => Ok(Response::new(RenameDataResponse {
success: false, success: false,
rename_data_resp: String::new(), rename_data_resp: String::new(),
rename_data_resp_bin: Vec::new().into(), rename_data_resp_bin: Vec::new().into(),
error: Some(err.into()), error: Some(err.into()),
})), })),
} },
} else { Err(err) => Ok(Response::new(RenameDataResponse {
Ok(Response::new(RenameDataResponse {
success: false, success: false,
rename_data_resp: String::new(), rename_data_resp: String::new(),
rename_data_resp_bin: Vec::new().into(), rename_data_resp_bin: Vec::new().into(),
error: Some(DiskError::other("cannot find disk".to_string()).into()), error: Some(err.into()),
})) })),
} }
} }
+8 -3
View File
@@ -377,10 +377,12 @@ pub(crate) mod timeout_wrapper_consumer {
} }
pub(crate) mod tonic_service_consumer { pub(crate) mod tonic_service_consumer {
#[cfg(test)]
pub(crate) use super::super::tonic_service::make_server;
#[cfg(test)] #[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::{heal_topology_fingerprint, make_heal_control_server_for_source};
pub(crate) use super::super::tonic_service::{ 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)] #[cfg(test)]
pub(crate) use rustfs_ecstore::api::storage::init_local_disks; pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
pub(crate) use rustfs_ecstore::api::storage::{ pub(crate) use rustfs_ecstore::api::storage::{
ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, all_local_disk_path, BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
find_local_disk_by_ref, init_local_disks_with_instance_ctx, init_lock_clients, 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, 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 DynReplicationPool = StorageReplicationPoolHandle;
pub(crate) type DynReader = ecstore_rio::DynReader; pub(crate) type DynReader = ecstore_rio::DynReader;
pub(crate) type ECStore = ecstore_storage::ECStore; 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; pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
#[cfg(test)] #[cfg(test)]
pub(crate) type Endpoints = ecstore_layout::Endpoints; pub(crate) type Endpoints = ecstore_layout::Endpoints;
+7 -1
View File
@@ -13,8 +13,14 @@
// limitations under the License. // 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_heal_control_server_with_cache;
pub(crate) use crate::storage::rpc::node_service::make_scanner_control_server;
#[cfg(test)] #[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::{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 use crate::storage::rpc::{make_heal_control_server, make_server, make_tier_mutation_control_server};
pub type NodeService = crate::storage::rpc::NodeService; 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,
};
+4 -1
View File
@@ -171,12 +171,15 @@ pub(crate) mod server {
} }
pub(crate) mod tonic_service { pub(crate) mod tonic_service {
#[cfg(test)]
pub(crate) use crate::storage::storage_api::tonic_service_consumer::make_server;
#[cfg(test)] #[cfg(test)]
pub(crate) use crate::storage::storage_api::tonic_service_consumer::{ pub(crate) use crate::storage::storage_api::tonic_service_consumer::{
heal_topology_fingerprint, make_heal_control_server_for_source, heal_topology_fingerprint, make_heal_control_server_for_source,
}; };
pub(crate) use crate::storage::storage_api::tonic_service_consumer::{ 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,
}; };
} }
} }
@@ -540,3 +540,530 @@ async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed
server_b.shutdown().await; server_b.shutdown().await;
server_a.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);
}
}