perf(protos): cache internode msgpack-only flag (#5191)

Cache the internode msgpack-only env gate after the first read so metadata RPC send paths avoid repeated environment parsing. Keep a reset hook for tests that intentionally change the env in-process.

Co-authored-by: heihutu <heihutu@gmail.com>

Cargo.lock

@@ -9748,6 +9748,7 @@ dependencies = [
  "rustfs-utils",
  "serde",
  "serde_json",
+ "temp-env",
  "tokio",
  "tonic",
  "tonic-prost",

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -3018,15 +3018,32 @@ mod tests {
 
     #[test]
     fn compat_json_dual_writes_by_default() {
-        let resp = sample_read_multiple_resp("file", b"data");
+        with_internode_msgpack_env(
-        let json = compat_json(&resp).expect("compat_json should encode");
+            [
-        assert!(!json.is_empty());
+                (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, None::<&str>),
-        assert_eq!(json, serde_json::to_string(&resp).expect("json should encode"));
+                (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>),
+            ],
+            || {
+                let resp = sample_read_multiple_resp("file", b"data");
+                let json = compat_json(&resp).expect("compat_json should encode");
+                assert!(!json.is_empty());
+                assert_eq!(json, serde_json::to_string(&resp).expect("json should encode"));
+            },
+        );
+    }
+
+    fn with_internode_msgpack_env<R>(vars: [(&'static str, Option<&'static str>); 2], f: impl FnOnce() -> R) -> R {
+        temp_env::with_vars(vars, || {
+            rustfs_protos::reset_internode_rpc_msgpack_only_cache();
+            let result = f();
+            rustfs_protos::reset_internode_rpc_msgpack_only_cache();
+            result
+        })
     }
 
     #[test]
     fn compat_json_keeps_json_when_msgpack_only_lacks_fleet_confirmation() {
-        temp_env::with_vars(
+        with_internode_msgpack_env(
             [
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")),
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>),
@@ -3043,7 +3060,7 @@ mod tests {
 
     #[test]
     fn compat_json_omits_json_only_after_fleet_confirmation() {
-        temp_env::with_vars(
+        with_internode_msgpack_env(
             [
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")),
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")),

crates/protos/Cargo.toml

@@ -54,3 +54,4 @@ doctest = false
 
 [dev-dependencies]
 serde_json = { workspace = true }
+temp-env = { workspace = true }

crates/protos/src/lib.rs

@@ -25,7 +25,7 @@ use std::{
     collections::HashMap,
     error::Error,
     sync::LazyLock,
-    sync::atomic::{AtomicUsize, Ordering},
+    sync::atomic::{AtomicU8, AtomicUsize, Ordering},
     time::{Duration, Instant},
 };
 use tokio::sync::Mutex;
@@ -54,7 +54,11 @@ pub const DEFAULT_GRPC_SERVER_MESSAGE_LEN: usize = 100 * 1024 * 1024;
 /// Default value: https://
 const RUSTFS_HTTPS_PREFIX: &str = "https://";
 const TLS_GENERATION_CACHE_MAX_SIZE: usize = 512;
+const INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET: u8 = 0;
+const INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE: u8 = 1;
+const INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE: u8 = 2;
 static TLS_GENERATION_CACHE: LazyLock<Mutex<HashMap<String, u64>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static INTERNODE_RPC_MSGPACK_ONLY_CACHE: AtomicU8 = AtomicU8::new(INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET);
 
 fn enforce_tls_generation_cache_bound(generation_cache: &mut HashMap<String, u64>, generation: u64, addr: &str) {
     if generation_cache.len() < TLS_GENERATION_CACHE_MAX_SIZE || generation_cache.contains_key(addr) {
@@ -642,13 +646,33 @@ mod tier_mutation_rpc_tests {
 /// decode the compatibility field. Operators must also set the fleet-confirmed guard after the
 /// convergence runbook proves every peer supports `_bin` and rollback.
 pub fn internode_rpc_msgpack_only() -> bool {
-    rustfs_utils::get_env_bool(
+    match INTERNODE_RPC_MSGPACK_ONLY_CACHE.load(Ordering::Acquire) {
+        INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE => return false,
+        INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE => return true,
+        _ => {}
+    }
+
+    let enabled = rustfs_utils::get_env_bool(
         rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY,
         rustfs_config::DEFAULT_INTERNODE_RPC_MSGPACK_ONLY,
     ) && rustfs_utils::get_env_bool(
         rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED,
         rustfs_config::DEFAULT_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED,
-    )
+    );
+    INTERNODE_RPC_MSGPACK_ONLY_CACHE.store(
+        if enabled {
+            INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE
+        } else {
+            INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE
+        },
+        Ordering::Release,
+    );
+    enabled
+}
+
+#[doc(hidden)]
+pub fn reset_internode_rpc_msgpack_only_cache() {
+    INTERNODE_RPC_MSGPACK_ONLY_CACHE.store(INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET, Ordering::Release);
 }
 
 /// Consecutive-failure threshold after which an internode peer is marked offline (grpc-optimization
@@ -951,6 +975,35 @@ mod tests {
         assert!(bulk_channel_pool_size() >= 1);
     }
 
+    #[test]
+    fn internode_rpc_msgpack_only_reuses_cached_env_until_reset() {
+        reset_internode_rpc_msgpack_only_cache();
+
+        temp_env::with_vars(
+            [
+                (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")),
+                (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")),
+            ],
+            || {
+                assert!(internode_rpc_msgpack_only());
+            },
+        );
+
+        temp_env::with_vars(
+            [
+                (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, None::<&str>),
+                (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>),
+            ],
+            || {
+                assert!(internode_rpc_msgpack_only(), "cached value should avoid repeated env reads");
+                reset_internode_rpc_msgpack_only_cache();
+                assert!(!internode_rpc_msgpack_only(), "reset must reload current env values");
+            },
+        );
+
+        reset_internode_rpc_msgpack_only_cache();
+    }
+
     #[tokio::test]
     async fn get_channel_for_class_bulk_reuses_control_cache_when_isolation_disabled() {
         // Isolation defaults off, so a Bulk request must reuse the control channel keyed by the

rustfs/src/storage/rpc/node_service/disk.rs

@@ -1215,9 +1215,18 @@ mod tests {
         );
     }
 
+    fn with_internode_msgpack_env<R>(vars: [(&'static str, Option<&'static str>); 2], f: impl FnOnce() -> R) -> R {
+        temp_env::with_vars(vars, || {
+            rustfs_protos::reset_internode_rpc_msgpack_only_cache();
+            let result = f();
+            rustfs_protos::reset_internode_rpc_msgpack_only_cache();
+            result
+        })
+    }
+
     #[test]
     fn compat_response_json_keeps_json_when_msgpack_only_lacks_fleet_confirmation() {
-        temp_env::with_vars(
+        with_internode_msgpack_env(
             [
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")),
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>),
@@ -1237,7 +1246,7 @@ mod tests {
 
     #[test]
     fn compat_response_json_omits_json_only_after_fleet_confirmation() {
-        temp_env::with_vars(
+        with_internode_msgpack_env(
             [
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")),
                 (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")),