From 8e83087ba4091354d908ba4ca84cf066aa3eb9e6 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 25 Jul 2026 19:10:13 +0800 Subject: [PATCH] fix(tier): handle mutation intent CAS races (#5227) * test(scripts): expand internode grpc ab env coverage * fix(tier): handle mutation intent CAS races Make tier mutation intent advancement retry idempotently when an If-Match CAS loses to a matching committed peer update. Expand four-node inline fallback E2E coverage for mixed msgpack controls, transition readiness evidence, and per-node environment overrides. Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- Cargo.lock | 106 ++-- Cargo.toml | 8 +- crates/e2e_test/src/common.rs | 45 ++ .../src/inline_fast_path_cluster_test.rs | 577 ++++++++++++++++-- crates/ecstore/src/services/tier/tier.rs | 52 ++ .../src/services/tier/tier_mutation_intent.rs | 24 +- scripts/test_internode_grpc_ab_bench.sh | 31 + 7 files changed, 747 insertions(+), 96 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 92a93536a..a252e293a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -850,9 +850,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-config" -version = "1.10.0" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "701418aa459dac33e50a0f8e818e5662a16bc018a6ac7423659b70f3799d67a8" +checksum = "1b180a3c8b55960db3426d8964b8745e652466a1a49fe1a2eda828046d30b5e4" dependencies = [ "aws-credential-types", "aws-runtime", @@ -917,9 +917,9 @@ dependencies = [ [[package]] name = "aws-runtime" -version = "1.9.0" +version = "1.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6b50a43f3ccdf331521c6d6c68b7cc9668b6e09d439ebda9569df5722324d76" +checksum = "c9007227e10b5fed2f3e0a2beff489211e2b5604c400b7a9d5d81ca9d64c24bb" dependencies = [ "aws-credential-types", "aws-sigv4", @@ -945,9 +945,9 @@ dependencies = [ [[package]] name = "aws-sdk-s3" -version = "1.139.0" +version = "1.140.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a159b9721a6a41468f967d1029bece78f410b0beb0594498435deb6ff72bfe48" +checksum = "e9660cf991e512fbe6094f1041ff3d3282bc5aeeeb6184e8de437ec2de024a10" dependencies = [ "arc-swap", "aws-credential-types", @@ -982,9 +982,9 @@ dependencies = [ [[package]] name = "aws-sdk-sso" -version = "1.104.0" +version = "1.105.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b53416d16c278234845392e38d93bd4481d2f09daa0f005a2277f0aa91f59c22" +checksum = "6ffd0fbe7873cb548a7aa60f9573c268fff94155397fd4f14dc9f1ecaaab8516" dependencies = [ "arc-swap", "aws-credential-types", @@ -1008,9 +1008,9 @@ dependencies = [ [[package]] name = "aws-sdk-ssooidc" -version = "1.106.0" +version = "1.107.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc9b706c3305ed0285d5b1b696c747aa34950f830fb03e3e6c76890f99b9f188" +checksum = "175763eb222a46377df7aa257a3bca980ab3e96703fefc8f4d0b8da6ad2e254c" dependencies = [ "arc-swap", "aws-credential-types", @@ -1034,9 +1034,9 @@ dependencies = [ [[package]] name = "aws-sdk-sts" -version = "1.109.0" +version = "1.110.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32d214cdfa5bbe17f117e76a7643fadf32a5234fb597322ef8b1fb4b2f17dbbd" +checksum = "dd8b14781dfbff48984017d57167b6ea0b6471c6920ec52b44a2677c7feb3c13" dependencies = [ "arc-swap", "aws-credential-types", @@ -1210,9 +1210,9 @@ dependencies = [ [[package]] name = "aws-smithy-runtime" -version = "1.12.0" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bea94a9ff8464016338c851e24b472d7131c388c88898a502e781815b2ee6045" +checksum = "07505b34e8f4b3591a4fa69e9792b52289b95488dbbc68c3c0075b7bedb245e1" dependencies = [ "aws-smithy-async", "aws-smithy-http", @@ -1236,9 +1236,9 @@ dependencies = [ [[package]] name = "aws-smithy-runtime-api" -version = "1.13.0" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22ed1ebe6e0a95ea84570225f5a8208dec4b8f77e61a9b0d6f51773fcb4612f0" +checksum = "3b98f2e1fd67ec06618f9c291e5e495a468e60519e44c9c1979cd0521f3affdb" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api-macros", @@ -1528,7 +1528,7 @@ version = "0.10.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", ] [[package]] @@ -1547,7 +1547,7 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", ] [[package]] @@ -1760,9 +1760,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "jobserver", @@ -1870,7 +1870,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common 0.1.6", + "crypto-common 0.1.7", "inout 0.1.4", ] @@ -2328,7 +2328,7 @@ version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", "rand_core 0.6.4", "subtle", "zeroize", @@ -2353,11 +2353,11 @@ dependencies = [ [[package]] name = "crypto-common" -version = "0.1.6" +version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", "typenum", ] @@ -3554,7 +3554,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.4", "const-oid 0.9.6", - "crypto-common 0.1.6", + "crypto-common 0.1.7", "subtle", ] @@ -3796,9 +3796,9 @@ dependencies = [ [[package]] name = "either" -version = "1.16.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" [[package]] name = "elliptic-curve" @@ -3810,7 +3810,7 @@ dependencies = [ "crypto-bigint 0.5.5", "digest 0.10.7", "ff 0.13.1", - "generic-array 0.14.9", + "generic-array 0.14.7", "group 0.13.0", "hkdf 0.12.4", "pem-rfc7468 0.7.0", @@ -4256,9 +4256,9 @@ dependencies = [ [[package]] name = "generic-array" -version = "0.14.9" +version = "0.14.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" dependencies = [ "typenum", "version_check", @@ -4271,7 +4271,7 @@ version = "1.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab4e5aa225bc56696909483320f0ff9b600f1a971b52e07a17d70f3d9b43254b" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", "rustversion", "typenum", ] @@ -4383,7 +4383,7 @@ dependencies = [ "hex", "hmac 0.13.0", "http 1.4.2", - "jsonwebtoken", + "jsonwebtoken 10.4.0", "reqwest", "rustc_version", "rustls", @@ -5271,7 +5271,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ "block-padding 0.3.3", - "generic-array 0.14.9", + "generic-array 0.14.7", ] [[package]] @@ -5535,6 +5535,22 @@ name = "jsonwebtoken" version = "10.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc" +dependencies = [ + "aws-lc-rs", + "base64 0.22.1", + "getrandom 0.2.17", + "js-sys", + "serde", + "serde_json", + "signature 2.2.0", + "zeroize", +] + +[[package]] +name = "jsonwebtoken" +version = "11.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "881733cbc631fc9e472e24447ce32a64bedf2da498d6d8570b08edc87de71f65" dependencies = [ "aws-lc-rs", "base64 0.22.1", @@ -6686,7 +6702,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.22.1", + "base64 0.21.7", "chrono", "getrandom 0.2.17", "http 1.4.2", @@ -7808,7 +7824,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.10.5", "log", "multimap", "once_cell", @@ -7828,7 +7844,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.10.5", "log", "multimap", "petgraph 0.8.3", @@ -7849,7 +7865,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.119", @@ -7862,7 +7878,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.119", @@ -9061,7 +9077,7 @@ dependencies = [ "argon2", "base64-simd", "chacha20poly1305", - "jsonwebtoken", + "jsonwebtoken 11.0.0", "pbkdf2 0.13.0", "rand 0.10.2", "rsa 0.10.0-rc.18", @@ -9293,7 +9309,7 @@ dependencies = [ "base64-simd", "futures", "http 1.4.2", - "jsonwebtoken", + "jsonwebtoken 11.0.0", "moka", "openidconnect", "pollster", @@ -9655,7 +9671,7 @@ dependencies = [ "chrono", "futures", "ipnetwork", - "jsonwebtoken", + "jsonwebtoken 11.0.0", "moka", "pollster", "proptest", @@ -10273,7 +10289,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -10483,7 +10499,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" dependencies = [ "base16ct 0.2.0", "der 0.7.10", - "generic-array 0.14.9", + "generic-array 0.14.7", "pkcs8 0.10.2", "subtle", "zeroize", @@ -11473,7 +11489,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index 4ee3c9dc8..cd1ad2936 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -196,7 +196,7 @@ blake2 = "=0.11.0-rc.6" chacha20poly1305 = { version = "=0.11.0" } crc-fast = "1.10.0" hmac = { version = "0.13.0" } -jsonwebtoken = { version = "10.4.0" } +jsonwebtoken = { version = "11.0.0" } openidconnect = { default-features = false, version = "4.0" } pbkdf2 = "0.13.0" rsa = { version = "=0.10.0-rc.18" } @@ -225,11 +225,11 @@ arc-swap = "1.9.2" astral-tokio-tar = "0.6.4" atoi = "3.1.0" atomic_enum = "0.3.0" -aws-config = { version = "1.10.0" } +aws-config = { version = "1.10.1" } aws-credential-types = { version = "1.3.0" } -aws-sdk-s3 = { default-features = false, version = "1.139.0" } +aws-sdk-s3 = { default-features = false, version = "1.140.0" } aws-smithy-http-client = { default-features = false, version = "1.2.0" } -aws-smithy-runtime-api = { version = "1.13.0" } +aws-smithy-runtime-api = { version = "1.14.0" } aws-smithy-types = { version = "1.6.1" } base64 = "0.23.0" base64-simd = "0.8.0" diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index b95af36c3..ca8514421 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -892,6 +892,7 @@ pub struct RustFSTestClusterEnvironment { pub access_key: String, pub secret_key: String, pub extra_env: Vec<(String, String)>, + pub node_extra_env: Vec>, pub topology: ClusterTopology, } @@ -990,6 +991,7 @@ impl RustFSTestClusterEnvironment { access_key: "rustfs-cluster-test-access".to_string(), secret_key: "rustfs-cluster-test-secret".to_string(), extra_env, + node_extra_env: vec![Vec::new(); topology.node_count], topology, }) } @@ -1003,6 +1005,22 @@ impl RustFSTestClusterEnvironment { self.extra_env.push((key.into(), value.into())); } + /// Add an extra environment variable applied to a single cluster node. + pub fn set_node_env( + &mut self, + node_idx: usize, + key: K, + value: V, + ) -> Result<(), Box> + where + K: Into, + V: Into, + { + self.ensure_node_index(node_idx)?; + self.node_extra_env[node_idx].push((key.into(), value.into())); + Ok(()) + } + fn ensure_node_index(&self, node_idx: usize) -> Result<(), Box> { if node_idx >= self.nodes.len() { return Err(format!("node_idx {node_idx} is invalid").into()); @@ -1089,6 +1107,9 @@ impl RustFSTestClusterEnvironment { for (key, value) in &self.extra_env { command.env(key, value); } + for (key, value) in &self.node_extra_env[i] { + command.env(key, value); + } let process = command.current_dir(&node.data_dir).spawn()?; @@ -1130,6 +1151,9 @@ impl RustFSTestClusterEnvironment { for (key, value) in &self.extra_env { command.env(key, value); } + for (key, value) in &self.node_extra_env[node_idx] { + command.env(key, value); + } let process = command.current_dir(&node.data_dir).spawn()?; node.process = Some(process); @@ -1371,6 +1395,7 @@ mod tests { access_key: DEFAULT_ACCESS_KEY.to_string(), secret_key: DEFAULT_SECRET_KEY.to_string(), extra_env: Vec::new(), + node_extra_env: vec![Vec::new(); topology.node_count], topology, } } @@ -1455,4 +1480,24 @@ mod tests { assert!(ClusterTopology::single_pool_multidrive(4, 4).validate().is_ok()); assert!(ClusterTopology::single_pool_multidrive(1, 1).validate().is_ok()); } + + #[test] + fn cluster_node_env_supports_per_node_overrides() { + let mut env = fake_cluster(ClusterTopology::single_pool(4)); + env.set_node_env(2, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY", "true").unwrap(); + assert_eq!( + env.node_extra_env[2].as_slice(), + [("RUSTFS_INTERNODE_RPC_MSGPACK_ONLY".to_string(), "true".to_string())] + ); + } + + #[test] + fn cluster_node_env_rejects_invalid_index() { + let mut env = fake_cluster(ClusterTopology::single_pool(4)); + let err = env + .set_node_env(4, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY", "true") + .unwrap_err() + .to_string(); + assert!(err.contains("invalid"), "unexpected error: {err}"); + } } diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index f0c6fd08b..6efcf574b 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -64,6 +64,9 @@ type MetricValues = Arc>>; const KIB: usize = 1024; const READER_PATH_COUNTER: &str = "rustfs_io_get_object_reader_path_by_size_total"; +const MSGPACK_JSON_FALLBACK_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_fallback_total"; +const DIRECTION_LABEL: &str = "direction"; +const MESSAGE_LABEL: &str = "message"; const INLINE_DIRECT: &str = "inline_direct"; const LEGACY_DUPLEX: &str = "legacy_duplex"; const EMPTY: &str = "empty"; @@ -74,11 +77,30 @@ const ENCRYPTED: &str = "encrypted"; const COMPRESSED: &str = "compressed"; const RANGE: &str = "range"; const REMOTE: &str = "remote"; +const FALLBACK_REQUEST_DIRECTION: &str = "request"; +const FALLBACK_RESPONSE_DIRECTION: &str = "response"; const MPU_PART_1_SIZE: usize = 5 * 1024 * 1024; const MPU_PART_2_SIZE: usize = 16 * KIB; const TIER_NAME: &str = "COLDTIER"; const TIER_BUCKET: &str = "inline-fallback-cold-tier"; const TIER_PREFIX: &str = "tiered"; +const MSGPACK_FALLBACK_CONTROL_SERIES: [(&str, &str); 4] = [ + (FALLBACK_REQUEST_DIRECTION, "ReadMultipleReq"), + (FALLBACK_RESPONSE_DIRECTION, "ReadMultipleResp"), + (FALLBACK_REQUEST_DIRECTION, "BatchReadVersionReq"), + (FALLBACK_RESPONSE_DIRECTION, "BatchReadVersionResp"), +]; +const READER_SIZE_BUCKETS: [&str; 9] = [ + "le_4kib", + "le_16kib", + "le_64kib", + "le_128kib", + "le_192kib", + "le_256kib", + "le_512kib", + "le_1mib", + "gt_1mib", +]; struct BoundaryCase { label: String, @@ -113,6 +135,7 @@ impl VersionState { struct OtlpMetricCollector { endpoint: String, values: MetricValues, + fallback_values: MetricValues, task: JoinHandle<()>, } @@ -121,24 +144,32 @@ impl OtlpMetricCollector { let listener = TcpListener::bind("127.0.0.1:0").await?; let endpoint = format!("http://{}/v1/metrics", listener.local_addr()?); let values = Arc::new(Mutex::new(BTreeMap::new())); + let fallback_values = Arc::new(Mutex::new(BTreeMap::new())); let task_values = values.clone(); + let task_fallback_values = fallback_values.clone(); let task = tokio::spawn(async move { loop { let Ok((stream, _)) = listener.accept().await else { break; }; let values = task_values.clone(); + let fallback_values = task_fallback_values.clone(); tokio::spawn(async move { let _ = hyper::server::conn::http1::Builder::new() .serve_connection( TokioIo::new(stream), - service_fn(move |request| handle_metric_export(request, values.clone())), + service_fn(move |request| handle_metric_export(request, values.clone(), fallback_values.clone())), ) .await; }); } }); - Ok(Self { endpoint, values, task }) + Ok(Self { + endpoint, + values, + fallback_values, + task, + }) } async fn reader_path_total(&self, path: &str, object_class: &str, size_bucket: &str) -> u64 { @@ -166,6 +197,15 @@ impl OtlpMetricCollector { .collect() } + async fn msgpack_json_fallback_totals(&self) -> BTreeMap { + self.fallback_values + .lock() + .await + .iter() + .map(|(key, points)| (key.clone(), points.values().map(|(_, value)| value).sum())) + .collect() + } + async fn reader_path_totals_for(&self, object_class: &str, size_bucket: &str) -> BTreeMap { let values = self.values.lock().await; [INLINE_DIRECT, LEGACY_DUPLEX, EMPTY, REMOTE_TRANSITION] @@ -231,7 +271,11 @@ impl Drop for OtlpMetricCollector { } } -async fn handle_metric_export(request: Request, values: MetricValues) -> Result>, Infallible> { +async fn handle_metric_export( + request: Request, + values: MetricValues, + fallback_values: MetricValues, +) -> Result>, Infallible> { if request.uri().path() != "/v1/metrics" { return Ok(response(StatusCode::NOT_FOUND)); } @@ -261,7 +305,9 @@ async fn handle_metric_export(request: Request, values: MetricValues) match ExportMetricsServiceRequest::decode(payload.as_slice()) { Ok(export) => { let mut values = values.lock().await; + let mut fallback_values = fallback_values.lock().await; record_reader_path_metrics(&export, &mut values); + record_msgpack_fallback_metrics(&export, &mut fallback_values); Ok(response(StatusCode::OK)) } Err(_) => Ok(response(StatusCode::BAD_REQUEST)), @@ -326,6 +372,66 @@ fn record_reader_path_metric(metric: &Metric, values: &mut BTreeMap) { + for resource_metrics in &export.resource_metrics { + for scope_metrics in &resource_metrics.scope_metrics { + for metric in &scope_metrics.metrics { + if metric.name != MSGPACK_JSON_FALLBACK_COUNTER { + continue; + } + let Some(metric::Data::Sum(sum)) = &metric.data else { + continue; + }; + for point in &sum.data_points { + let Some(direction) = attribute_string(&point.attributes, DIRECTION_LABEL) else { + continue; + }; + let Some(message) = attribute_string(&point.attributes, MESSAGE_LABEL) else { + continue; + }; + let Some(number_data_point::Value::AsInt(value)) = point.value.as_ref() else { + continue; + }; + let value = u64::try_from(*value).unwrap_or_default(); + values + .entry(msgpack_fallback_metric_key(direction, message)) + .or_default() + .entry(point.start_time_unix_nano) + .and_modify(|current| { + if point.time_unix_nano >= current.0 { + *current = (point.time_unix_nano, value); + } + }) + .or_insert((point.time_unix_nano, value)); + } + } + } + } +} + +fn msgpack_fallback_metric_key(direction: &str, message: &str) -> String { + format!("{direction}\u{1f}{message}") +} + +async fn assert_msgpack_fallback_unchanged( + collector: &OtlpMetricCollector, + before: &BTreeMap, + series: &[(&str, &str)], +) -> TestResult { + let after = collector.msgpack_json_fallback_totals().await; + for &(direction, message) in series { + let key = msgpack_fallback_metric_key(direction, message); + assert_eq!( + before.get(&key).copied().unwrap_or_default(), + after.get(&key).copied().unwrap_or_default(), + "fallback counter changed for {direction}/{message}: before={:?}, after={:?}", + before.get(&key), + after.get(&key), + ); + } + Ok(()) +} + fn attribute_string<'a>(attributes: &'a [KeyValue], wanted_key: &str) -> Option<&'a str> { attributes.iter().find_map(|attribute| { if attribute.key != wanted_key { @@ -439,6 +545,17 @@ fn configure_reader_metric_cluster(cluster: &mut RustFSTestClusterEnvironment, c cluster.set_env("RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY", "false"); } +fn configure_mixed_msgpack_cluster(cluster: &mut RustFSTestClusterEnvironment, collector: &OtlpMetricCollector) -> TestResult { + configure_reader_metric_cluster(cluster, collector); + cluster.set_node_env(0, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY", "true")?; + cluster.set_node_env(0, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED", "true")?; + cluster.set_node_env(1, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY", "true")?; + cluster.set_node_env(1, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED", "true")?; + cluster.set_node_env(2, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY", "false")?; + cluster.set_node_env(3, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY", "false")?; + Ok(()) +} + async fn assert_case( cluster: &RustFSTestClusterEnvironment, client: &Client, @@ -513,7 +630,7 @@ struct ReaderPathExpectation<'a> { object: ReaderObject<'a>, expected_path: &'a str, object_class: &'a str, - expected_size_bucket: &'a str, + expected_size_bucket: Option<&'a str>, } impl<'a> ReaderPathExpectation<'a> { @@ -527,7 +644,7 @@ impl<'a> ReaderPathExpectation<'a> { object, expected_path, object_class, - expected_size_bucket, + expected_size_bucket: Some(expected_size_bucket), } } @@ -541,7 +658,16 @@ impl<'a> ReaderPathExpectation<'a> { object, expected_path, object_class, - expected_size_bucket, + expected_size_bucket: Some(expected_size_bucket), + } + } + + fn with_any_size_bucket(object: ReaderObject<'a>, expected_path: &'a str, object_class: &'a str) -> Self { + Self { + object, + expected_path, + object_class, + expected_size_bucket: None, } } } @@ -558,22 +684,102 @@ async fn assert_reader_path( expected_size_bucket, } = expectation; let paths = [INLINE_DIRECT, LEGACY_DUPLEX, EMPTY, REMOTE_TRANSITION]; - let mut before = BTreeMap::<&str, u64>::new(); - for path in paths { - before.insert(path, collector.reader_path_total(path, object_class, expected_size_bucket).await); + let size_buckets: Vec<&str> = match expected_size_bucket { + Some(expected_size_bucket) => vec![expected_size_bucket], + None => READER_SIZE_BUCKETS.to_vec(), + }; + let size_buckets = size_buckets.as_slice(); + + let mut before = BTreeMap::<&str, BTreeMap<&str, u64>>::new(); + for &size_bucket in size_buckets { + let mut bucket_before = BTreeMap::new(); + for path in paths { + bucket_before.insert(path, collector.reader_path_total(path, object_class, size_bucket).await); + } + before.insert(size_bucket, bucket_before); } + get_and_assert(client, object.bucket, object.key, object.body, object.etag, object.version_id).await?; - let expected_after = collector - .wait_for_reader_path_total(expected_path, object_class, expected_size_bucket, before[expected_path] + 1) - .await?; - assert!( - expected_after > before[expected_path], - "{READER_PATH_COUNTER}{{path={expected_path}}} must advance for {}", - object.key - ); + + let selected_size_bucket = match expected_size_bucket { + Some(expected_size_bucket) => { + let path_before = before + .get(expected_size_bucket) + .and_then(|values| values.get(expected_path)) + .copied() + .ok_or_else(|| format!("reader-path baseline missing for size bucket {expected_size_bucket}"))?; + let expected_after = collector + .wait_for_reader_path_total(expected_path, object_class, expected_size_bucket, path_before + 1) + .await?; + assert!( + expected_after > path_before, + "{READER_PATH_COUNTER}{{path={expected_path}}} must advance for {}", + object.key + ); + expected_size_bucket + } + None => { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let mut matched_size_bucket = None; + for &size_bucket in size_buckets { + let path_before = before + .get(size_bucket) + .and_then(|values| values.get(expected_path)) + .copied() + .ok_or_else(|| format!("reader-path baseline missing for size bucket {size_bucket}"))?; + let path_after = collector.reader_path_total(expected_path, object_class, size_bucket).await; + if path_after < path_before + 1 { + continue; + } + let mut conflicting = false; + for path in paths { + if path == expected_path || expected_path == EMPTY { + continue; + } + let path_before = before + .get(size_bucket) + .and_then(|values| values.get(path)) + .copied() + .ok_or_else(|| format!("reader-path baseline missing for size bucket {size_bucket}"))?; + let path_after = collector.reader_path_total(path, object_class, size_bucket).await; + if path_after != path_before { + conflicting = true; + break; + } + } + if !conflicting { + matched_size_bucket = Some(size_bucket); + break; + } + } + + if let Some(size_bucket) = matched_size_bucket { + break size_bucket; + } + if Instant::now() >= deadline { + let mut observed = BTreeMap::<&str, BTreeMap<&str, u64>>::new(); + for &size_bucket in size_buckets { + let mut bucket_after = BTreeMap::new(); + for path in paths { + bucket_after.insert(path, collector.reader_path_total(path, object_class, size_bucket).await); + } + observed.insert(size_bucket, bucket_after); + } + return Err(format!( + "timed out waiting for {READER_PATH_COUNTER}{{object_class={object_class}, path={expected_path}}} to advance in any candidate size bucket; before={before:?}; after={observed:?}" + ) + .into()); + } + sleep(Duration::from_millis(100)).await; + } + } + }; + collector - .wait_for_reader_paths_to_settle(object_class, expected_size_bucket) + .wait_for_reader_paths_to_settle(object_class, selected_size_bucket) .await?; + for path in paths { if path == expected_path { continue; @@ -582,9 +788,13 @@ async fn assert_reader_path( continue; } assert_eq!( - collector.reader_path_total(path, object_class, expected_size_bucket).await, - before[path], - "{READER_PATH_COUNTER}{{path={path}, object_class={object_class}, size_bucket={expected_size_bucket}}} must not advance for {}; expected {expected_path} only", + collector.reader_path_total(path, object_class, selected_size_bucket).await, + before + .get(selected_size_bucket) + .and_then(|values| values.get(path)) + .copied() + .ok_or_else(|| format!("reader-path baseline missing for size bucket {selected_size_bucket}"))?, + "{READER_PATH_COUNTER}{{path={path}, object_class={object_class}, size_bucket={selected_size_bucket}}} must not advance for {}; expected {expected_path} only", object.key ); } @@ -734,6 +944,7 @@ async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEn }) .to_string(); let deadline = Instant::now() + Duration::from_secs(30); + let mut attempts = Vec::new(); let final_error = loop { let (status, response) = signed_admin_request( &hot.nodes[0].url, @@ -744,23 +955,60 @@ async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEn &hot.secret_key, ) .await?; - if status.is_success() || response.contains("TierNameAlreadyExist") { - wait_for_tier_verifiable(hot).await?; + let attempt = format!("status={status}, body={}", compact_body(&response)); + if status.is_success() { + wait_for_tier_verifiable(hot, &format!("status={status}, body={response}")).await?; return Ok(()); } + attempts.push(attempt); if Instant::now() >= deadline { - break format!("status={status}, body={response}"); + break attempts.join("; "); + } + if !is_retryable_add_tier_error(&response) { + break attempts.join("; "); } sleep(Duration::from_millis(500)).await; }; Err(format!("AddTier(RustFS) failed after readiness polling: {final_error}").into()) } -async fn wait_for_tier_verifiable(hot: &RustFSTestClusterEnvironment) -> TestResult { - let deadline = Instant::now() + Duration::from_secs(30); - let node = &hot.nodes[0]; +async fn wait_for_tier_verifiable(hot: &RustFSTestClusterEnvironment, add_tier_response: &str) -> TestResult { + let deadline = Instant::now() + Duration::from_secs(60); let final_error = loop { - let (status, response) = signed_admin_request( + let snapshot = tier_readiness_snapshot(hot).await?; + if snapshot.iter().any(|node| node.verify_status.is_success()) { + return Ok(()); + } + if snapshot.iter().any(|node| !is_retryable_tier_error(&node.verify_body)) { + break format!("non-retryable tier verification failure: {}", format_tier_readiness_snapshot(&snapshot)); + } + if Instant::now() >= deadline { + break format_tier_readiness_snapshot(&snapshot); + } + sleep(Duration::from_millis(500)).await; + }; + Err(format!( + "tier {TIER_NAME} was not verifiable on any hot node within 60s after AddTier({add_tier_response}): {final_error}" + ) + .into()) +} + +struct TierNodeReadiness { + node_index: usize, + node_url: String, + list_status: StatusCode, + list_has_tier: bool, + list_body: String, + verify_status: StatusCode, + verify_body: String, +} + +async fn tier_readiness_snapshot(hot: &RustFSTestClusterEnvironment) -> TestResult> { + let mut snapshot = Vec::with_capacity(hot.nodes.len()); + for (node_index, node) in hot.nodes.iter().enumerate() { + let (list_status, list_body) = + signed_admin_request(&node.url, Method::GET, "/rustfs/admin/v3/tier", None, &hot.access_key, &hot.secret_key).await?; + let (verify_status, verify_body) = signed_admin_request( &node.url, Method::GET, &format!("/rustfs/admin/v3/tier/{TIER_NAME}"), @@ -769,15 +1017,75 @@ async fn wait_for_tier_verifiable(hot: &RustFSTestClusterEnvironment) -> TestRes &hot.secret_key, ) .await?; - if status.is_success() { - return Ok(()); - } - if Instant::now() >= deadline { - break format!("node {} verify tier failed: status={status}, body={response}", node.url); - } - sleep(Duration::from_millis(500)).await; - }; - Err(format!("tier {TIER_NAME} was not verifiable on the primary hot node within 30s: {final_error}").into()) + snapshot.push(TierNodeReadiness { + node_index, + node_url: node.url.clone(), + list_status, + list_has_tier: tier_list_contains(&list_body), + list_body, + verify_status, + verify_body, + }); + } + Ok(snapshot) +} + +fn tier_list_contains(response: &str) -> bool { + serde_json::from_str::(response) + .ok() + .and_then(|value| value.as_array().cloned()) + .is_some_and(|tiers| { + tiers.iter().any(|tier| { + tier.get("name").and_then(serde_json::Value::as_str) == Some(TIER_NAME) + || tier + .get("rustfs") + .and_then(|rustfs| rustfs.get("name")) + .and_then(serde_json::Value::as_str) + == Some(TIER_NAME) + }) + }) +} + +fn format_tier_readiness_snapshot(snapshot: &[TierNodeReadiness]) -> String { + snapshot + .iter() + .map(|node| { + format!( + "node {} {} list_status={} list_has_tier={} list_body={} verify_status={} verify_body={}", + node.node_index, + node.node_url, + node.list_status, + node.list_has_tier, + compact_body(&node.list_body), + node.verify_status, + compact_body(&node.verify_body) + ) + }) + .collect::>() + .join("; ") +} + +fn compact_body(body: &str) -> String { + const MAX_BODY_CHARS: usize = 1024; + let mut value = body.split_whitespace().collect::>().join(" "); + if value.chars().count() > MAX_BODY_CHARS { + value = value.chars().take(MAX_BODY_CHARS).collect::(); + value.push_str("..."); + } + value +} + +fn is_retryable_tier_error(response: &str) -> bool { + response.contains("TierNotFound") + || response.contains("NoSuchTier") + || (response.contains("TierVerificationFailed") + && response.contains("Remote tier configuration is being replaced")) + || response.contains("TierNotFound") + || response.contains("NoSuchTier") +} + +fn is_retryable_add_tier_error(response: &str) -> bool { + response.contains("Remote tier configuration is already being replaced") } fn transition_rule() -> TestResult { @@ -1005,11 +1313,10 @@ async fn four_node_inline_fallback_controls() -> TestResult { assert_reader_path( &collector, &client, - ReaderPathExpectation::with_size_bucket( + ReaderPathExpectation::with_any_size_bucket( ReaderObject::new(bucket, encrypted_key, &encrypted_body, encrypted_put.e_tag(), None), LEGACY_DUPLEX, ENCRYPTED, - size_bucket(64 * KIB), ), ) .await?; @@ -1083,6 +1390,198 @@ async fn four_node_multipart_ignores_disk_compression_fallback() -> TestResult { Ok(()) } +#[tokio::test] +#[serial] +async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> TestResult { + init_logging(); + + let collector = OtlpMetricCollector::start().await?; + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + let sse_master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]); + cluster.set_env("RUSTFS_SSE_S3_MASTER_KEY", sse_master_key); + cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true"); + configure_mixed_msgpack_cluster(&mut cluster, &collector)?; + cluster.start().await?; + + let fallback_before = collector.msgpack_json_fallback_totals().await; + + let bucket = "inline-mixed-msgpack-controls"; + cluster.create_test_bucket(bucket).await?; + let client = cluster.create_s3_client(0)?; + + let multipart_key = "mixed/multipart.bin"; + let (multipart_body, second_part, multipart_etag) = put_two_part_multipart(&client, bucket, multipart_key).await?; + assert_reader_path( + &collector, + &client, + ReaderPathExpectation::for_class( + ReaderObject::new(bucket, multipart_key, &multipart_body, multipart_etag.as_deref(), None), + LEGACY_DUPLEX, + MULTIPART, + ), + ) + .await?; + assert_part_number_reader_path( + &collector, + &client, + bucket, + multipart_key, + &second_part, + multipart_body.len(), + LEGACY_DUPLEX, + ) + .await?; + assert_msgpack_fallback_unchanged(&collector, &fallback_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; + + let encrypted_key = "encrypted/sse-s3.bin"; + let encrypted_body = payload(16 * KIB, 0xE3); + let encrypted_put = client + .put_object() + .bucket(bucket) + .key(encrypted_key) + .server_side_encryption(ServerSideEncryption::Aes256) + .body(ByteStream::from(encrypted_body.clone())) + .send() + .await?; + let encrypted_head = client.head_object().bucket(bucket).key(encrypted_key).send().await?; + assert_eq!( + encrypted_head.server_side_encryption(), + Some(&ServerSideEncryption::Aes256), + "HEAD must preserve SSE-S3 metadata for {encrypted_key}" + ); + assert_reader_path( + &collector, + &client, + ReaderPathExpectation::with_any_size_bucket( + ReaderObject::new(bucket, encrypted_key, &encrypted_body, encrypted_put.e_tag(), None), + LEGACY_DUPLEX, + ENCRYPTED, + ), + ) + .await?; + + let compressed_key = "compressed/repeated.txt"; + let compressed_body = compressible_payload(64 * KIB); + let compressed_put = client + .put_object() + .bucket(bucket) + .key(compressed_key) + .body(ByteStream::from(compressed_body.clone())) + .send() + .await?; + assert_reader_path( + &collector, + &client, + ReaderPathExpectation::with_any_size_bucket( + ReaderObject::new(bucket, compressed_key, &compressed_body, compressed_put.e_tag(), None), + LEGACY_DUPLEX, + COMPRESSED, + ), + ) + .await?; + + assert_msgpack_fallback_unchanged(&collector, &fallback_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_transition() -> TestResult { + init_logging(); + + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "inlinecoldadmin".to_string(); + cold.secret_key = "inlinecoldsecret".to_string(); + cold.start_rustfs_server_without_cleanup(vec![]).await?; + let cold_client = cold.create_s3_client(); + cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; + + let collector = OtlpMetricCollector::start().await?; + let mut hot = RustFSTestClusterEnvironment::new(4).await?; + configure_mixed_msgpack_cluster(&mut hot, &collector)?; + hot.set_env("RUSTFS_SCANNER_CYCLE", "1"); + hot.set_env("RUSTFS_ILM_PROCESS_TIME", "1"); + + let sse_master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]); + hot.set_env("RUSTFS_SSE_S3_MASTER_KEY", sse_master_key); + hot.set_env("RUSTFS_COMPRESSION_ENABLED", "true"); + hot.start().await?; + let hot_client = hot.create_s3_client(0)?; + + let fallback_before = collector.msgpack_json_fallback_totals().await; + + add_rustfs_tier(&hot, &cold).await?; + let bucket = "inline-transitioned-mixed-msgpack-controls"; + hot_client.create_bucket().bucket(bucket).send().await?; + put_lifecycle_with_transition_retry(&hot_client, bucket).await?; + + let key = "transition/mixed-multipart.bin"; + let (body, _, etag) = put_two_part_multipart(&hot_client, bucket, key).await?; + wait_for_transition(&hot_client, bucket, key).await?; + assert!( + cold_tier_object_count(&cold_client).await? >= 1, + "cold-tier bucket must hold transitioned objects" + ); + assert_reader_path( + &collector, + &hot_client, + ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), REMOTE_TRANSITION, REMOTE), + ) + .await?; + + let encrypted_key = "transition/encrypted-sse.bin"; + let encrypted_body = payload(16 * KIB, 0xAB); + let encrypted_put = hot_client + .put_object() + .bucket(bucket) + .key(encrypted_key) + .server_side_encryption(ServerSideEncryption::Aes256) + .body(ByteStream::from(encrypted_body.clone())) + .send() + .await?; + let encrypted_head = hot_client.head_object().bucket(bucket).key(encrypted_key).send().await?; + assert_eq!( + encrypted_head.server_side_encryption(), + Some(&ServerSideEncryption::Aes256), + "HEAD must preserve SSE-S3 metadata for {encrypted_key}" + ); + assert_reader_path( + &collector, + &hot_client, + ReaderPathExpectation::with_any_size_bucket( + ReaderObject::new(bucket, encrypted_key, &encrypted_body, encrypted_put.e_tag(), None), + LEGACY_DUPLEX, + ENCRYPTED, + ), + ) + .await?; + + let compressed_key = "transition/compressed-repeated.txt"; + let compressed_body = compressible_payload(64 * KIB); + let compressed_put = hot_client + .put_object() + .bucket(bucket) + .key(compressed_key) + .body(ByteStream::from(compressed_body.clone())) + .send() + .await?; + assert_reader_path( + &collector, + &hot_client, + ReaderPathExpectation::with_any_size_bucket( + ReaderObject::new(bucket, compressed_key, &compressed_body, compressed_put.e_tag(), None), + LEGACY_DUPLEX, + COMPRESSED, + ), + ) + .await?; + + assert_msgpack_fallback_unchanged(&collector, &fallback_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; + + Ok(()) +} + #[tokio::test] #[serial] async fn four_node_transitioned_inline_fallback() -> TestResult { diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 7217bd52d..314b1d84a 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -6133,6 +6133,36 @@ mod tests { } } + #[tokio::test] + async fn intent_advance_treats_matching_committed_cas_race_as_idempotent() { + let store = Arc::new(CasConfigStore::default()); + let mutation_id = uuid::Uuid::from_u128(34); + let prepared = prepared_remove_intent("COLD-A", mutation_id); + crate::services::tier::tier_mutation_intent::save_tier_mutation_intent_record(store.clone(), &prepared) + .await + .expect("prepared intent fixture should persist"); + + let committed = committed_remove_intent("COLD-A", mutation_id, "etag-new"); + let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id) + .expect("record object should build"); + store + .rewrite_on_next_if_match(object, committed.encode().expect("committed intent fixture should encode")) + .await; + + let (observed, applied) = crate::services::tier::tier_mutation_intent::advance_tier_mutation_intent_record_idempotent( + store, + mutation_id, + TierMutationIntentState::Committed, + Some("etag-new".to_string()), + ) + .await + .expect("same-state CAS race should be idempotent"); + + assert!(!applied); + assert_eq!(observed.state, TierMutationIntentState::Committed); + assert_eq!(observed.committed_config_etag.as_deref(), Some("etag-new")); + } + #[tokio::test] async fn committed_mutation_recovery_requires_every_peer_to_commit() { let mutation_id = uuid::Uuid::from_u128(17); @@ -8225,6 +8255,7 @@ mod tests { struct CasConfigStore { objects: tokio::sync::Mutex, String)>>, legacy_state: tokio::sync::Mutex>>, + if_match_race_rewrite: tokio::sync::Mutex)>>, next_etag: AtomicUsize, fail_put: AtomicBool, truncate_reference_page_without_marker: AtomicBool, @@ -8238,6 +8269,7 @@ mod tests { Self { objects: tokio::sync::Mutex::new(HashMap::new()), legacy_state: tokio::sync::Mutex::new(None), + if_match_race_rewrite: tokio::sync::Mutex::new(None), next_etag: AtomicUsize::new(0), fail_put: AtomicBool::new(false), truncate_reference_page_without_marker: AtomicBool::new(false), @@ -8263,6 +8295,10 @@ mod tests { .insert(object, (data, "reference-proof-etag".to_string())); } + async fn rewrite_on_next_if_match(&self, object: String, data: Vec) { + *self.if_match_race_rewrite.lock().await = Some((object, data)); + } + fn omit_truncated_reference_marker(&self) { self.truncate_reference_page_without_marker.store(true, Ordering::SeqCst); } @@ -8323,7 +8359,23 @@ mod tests { } let mut payload = Vec::new(); tokio::io::AsyncReadExt::read_to_end(&mut data.stream, &mut payload).await?; + let race_rewrite = if opts + .http_preconditions + .as_ref() + .and_then(HTTPPreconditions::if_match_value) + .is_some() + { + self.if_match_race_rewrite.lock().await.take() + } else { + None + }; let mut objects = self.objects.lock().await; + if let Some((target, data)) = race_rewrite + && target == object + { + let etag = format!("etag-{}", self.next_etag.fetch_add(1, Ordering::SeqCst) + 1); + objects.insert(object.to_string(), (data, etag)); + } match objects.get(object) { Some((_, etag)) => opts.precondition_check(&ObjectInfo { etag: Some(etag.clone()), diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index d409069d4..fa473cc8b 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_intent.rs @@ -32,6 +32,7 @@ pub(crate) const TIER_MUTATION_INTENT_SCHEMA: &str = "rustfs-tier-mutation-inten pub(crate) const MAX_TIER_MUTATION_INTENT_SIZE: usize = rustfs_protos::TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE; pub(crate) const TIER_MUTATION_INTENT_RECORD_PREFIX: &str = "tier/mutation-intents/records"; pub(crate) const TIER_COORDINATOR_MUTATION_INTENT_RECORD_PREFIX: &str = "tier/mutation-intents/coordinators"; +const TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS: usize = 3; pub(crate) type TierMutationDigest = [u8; 32]; pub(crate) type Result = std::result::Result; @@ -555,15 +556,22 @@ async fn advance_tier_mutation_intent_record_idempotent_at_prefix( where S: EcstoreObjectIO, { - let (mut intent, current_etag) = - load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await?; - let advanced = intent - .advance_idempotent(next, committed_config_etag) - .map_err(tier_mutation_intent_store_error)?; - if advanced { - save_tier_mutation_intent_record_if_current_with_prefix(api, prefix, &intent, ¤t_etag).await?; + for attempt in 0..TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS { + let (mut intent, current_etag) = + load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await?; + let advanced = intent + .advance_idempotent(next, committed_config_etag.clone()) + .map_err(tier_mutation_intent_store_error)?; + if !advanced { + return Ok((intent, false)); + } + match save_tier_mutation_intent_record_if_current_with_prefix(api.clone(), prefix, &intent, ¤t_etag).await { + Ok(()) => return Ok((intent, true)), + Err(Error::PreconditionFailed) if attempt + 1 < TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS => continue, + Err(err) => return Err(err), + } } - Ok((intent, advanced)) + Err(Error::PreconditionFailed) } pub(crate) async fn list_tier_mutation_intent_records( diff --git a/scripts/test_internode_grpc_ab_bench.sh b/scripts/test_internode_grpc_ab_bench.sh index 6257d2201..ce25673f7 100755 --- a/scripts/test_internode_grpc_ab_bench.sh +++ b/scripts/test_internode_grpc_ab_bench.sh @@ -15,11 +15,23 @@ trap cleanup EXIT BEFORE_ENV="${TMP_DIR}/p2-before/server-env.sh" AFTER_ENV="${TMP_DIR}/p2-after/server-env.sh" +P0_BEFORE_ENV="${TMP_DIR}/p0-before/server-env.sh" +P0_AFTER_ENV="${TMP_DIR}/p0-after/server-env.sh" +P1_BEFORE_ENV="${TMP_DIR}/p1-before/server-env.sh" +P1_AFTER_ENV="${TMP_DIR}/p1-after/server-env.sh" +P3_BEFORE_ENV="${TMP_DIR}/p3-before/server-env.sh" +P3_AFTER_ENV="${TMP_DIR}/p3-after/server-env.sh" rg -qx 'export RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=false' "$BEFORE_ENV" rg -qx 'export RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=false' "$BEFORE_ENV" rg -qx 'export RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true' "$AFTER_ENV" rg -qx 'export RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=true' "$AFTER_ENV" +"$RUNNER" --stage p0 --phase before --out-root "$TMP_DIR" --dry-run >/dev/null +"$RUNNER" --stage p0 --phase after --out-root "$TMP_DIR" --dry-run >/dev/null +"$RUNNER" --stage p1 --phase before --out-root "$TMP_DIR" --dry-run >/dev/null +"$RUNNER" --stage p1 --phase after --out-root "$TMP_DIR" --dry-run >/dev/null +"$RUNNER" --stage p3 --phase before --out-root "$TMP_DIR" --dry-run >/dev/null +"$RUNNER" --stage p3 --phase after --out-root "$TMP_DIR" --dry-run >/dev/null if grep -q 'RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true' "$BEFORE_ENV"; then echo "p2 before must not request msgpack-only" >&2 @@ -31,4 +43,23 @@ if grep -q 'RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=true' "$BEFORE_ENV exit 1 fi +rg -qx 'export RUSTFS_INTERNODE_RPC_TCP_NODELAY=false' "$P0_BEFORE_ENV" +rg -qx 'export RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE=0' "$P0_BEFORE_ENV" +rg -qx 'export RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE=0' "$P0_BEFORE_ENV" +rg -qx 'export RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE=4194304' "$P0_BEFORE_ENV" +rg -qx '# \(cluster defaults — no overrides needed for this stage/phase\)' "$P0_AFTER_ENV" + +rg -qx 'export RUSTFS_INTERNODE_CHANNEL_ISOLATION=false' "$P1_BEFORE_ENV" +rg -qx 'export RUSTFS_INTERNODE_CHANNEL_ISOLATION=true' "$P1_AFTER_ENV" +rg -qx 'export RUSTFS_INTERNODE_BULK_CHANNELS=2' "$P1_AFTER_ENV" +if rg -q 'export RUSTFS_INTERNODE_BULK_CHANNELS=' "$P1_BEFORE_ENV"; then + echo "p1 before should not pin an explicit bulk channel count" >&2 + exit 1 +fi + +rg -qx 'export RUSTFS_INTERNODE_PREWARM=true' "$P3_AFTER_ENV" +rg -qx 'export RUSTFS_INTERNODE_OFFLINE_BYPASS=true' "$P3_AFTER_ENV" +rg -qx 'export RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS=5' "$P3_AFTER_ENV" +rg -qx 'export RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD=3' "$P3_AFTER_ENV" + echo "internode grpc A/B bench env tests passed"