fix(data-usage): make empty checks exhaustive (#5424)

crates/concurrency/Cargo.toml

@@ -10,6 +10,9 @@ description = "Shared concurrency contract types for RustFS - workload admission
 keywords = ["rustfs", "concurrency", "admission", "backpressure", "workers"]
 categories = ["concurrency", "filesystem"]
 
+[lints]
+workspace = true
+
 [dependencies]
 # Internal crates
 rustfs-io-core = { workspace = true }

crates/data-usage/src/data_usage.rs

@@ -507,16 +507,29 @@ pub struct ReplicationStats {
 
 impl ReplicationStats {
     pub fn is_empty(&self) -> bool {
-        self.pending_size == 0
-            && self.replicated_size == 0
-            && self.failed_size == 0
-            && self.failed_count == 0
-            && self.pending_count == 0
-            && self.missed_threshold_size == 0
-            && self.after_threshold_size == 0
-            && self.missed_threshold_count == 0
-            && self.after_threshold_count == 0
-            && self.replicated_count == 0
+        let Self {
+            pending_size,
+            replicated_size,
+            failed_size,
+            failed_count,
+            pending_count,
+            missed_threshold_size,
+            after_threshold_size,
+            missed_threshold_count,
+            after_threshold_count,
+            replicated_count,
+        } = self;
+
+        *pending_size == 0
+            && *replicated_size == 0
+            && *failed_size == 0
+            && *failed_count == 0
+            && *pending_count == 0
+            && *missed_threshold_size == 0
+            && *after_threshold_size == 0
+            && *missed_threshold_count == 0
+            && *after_threshold_count == 0
+            && *replicated_count == 0
     }
 
     #[deprecated(note = "use is_empty instead")]
@@ -535,7 +548,13 @@ pub struct ReplicationAllStats {
 
 impl ReplicationAllStats {
     pub fn is_empty(&self) -> bool {
-        self.replica_size == 0 && self.replica_count == 0 && self.targets.values().all(ReplicationStats::is_empty)
+        let Self {
+            replica_size,
+            replica_count,
+            targets,
+        } = self;
+
+        *replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
     }
 
     #[deprecated(note = "use is_empty instead")]
@@ -1594,7 +1613,6 @@ mod tests {
     }
 
     #[test]
-    #[allow(deprecated)]
     fn replication_stats_empty_checks_every_field() {
         type SetField = fn(&mut ReplicationStats);
 
@@ -1611,16 +1629,15 @@ mod tests {
             ("replicated_count", |stats| stats.replicated_count = 1),
         ];
 
-        assert!(ReplicationStats::default().empty());
+        assert!(ReplicationStats::default().is_empty());
         for (field, set_nonzero) in cases {
             let mut stats = ReplicationStats::default();
             set_nonzero(&mut stats);
-            assert!(!stats.empty(), "{field} must make replication stats non-empty");
+            assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
         }
     }
 
     #[test]
-    #[allow(deprecated)]
     fn replication_all_stats_empty_checks_aggregate_fields_independently() {
         let cases = [
             (
@@ -1639,16 +1656,16 @@ mod tests {
             ),
         ];
 
-        assert!(ReplicationAllStats::default().empty());
+        assert!(ReplicationAllStats::default().is_empty());
         for (field, stats) in cases {
-            assert!(!stats.empty(), "{field} must make aggregate replication stats non-empty");
+            assert!(!stats.is_empty(), "{field} must make aggregate replication stats non-empty");
         }
 
         let empty_targets = ReplicationAllStats {
             targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
             ..Default::default()
         };
-        assert!(empty_targets.empty(), "all-empty targets must keep aggregate stats empty");
+        assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
 
         let stats = ReplicationAllStats {
             targets: HashMap::from([
@@ -1663,7 +1680,7 @@ mod tests {
             ]),
             ..Default::default()
         };
-        assert!(!stats.empty(), "a non-empty target must make aggregate replication stats non-empty");
+        assert!(!stats.is_empty(), "a non-empty target must make aggregate replication stats non-empty");
     }
 
     #[test]

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

@@ -443,6 +443,16 @@ pub fn set_tonic_canonical_body_digest<T>(request: &mut tonic::Request<T>, canon
     Ok(())
 }
 
+pub fn set_tonic_mutation_body_digest<T: rustfs_protos::CanonicalMutationBody>(
+    request: &mut tonic::Request<T>,
+) -> std::io::Result<()> {
+    let canonical_body = request
+        .get_ref()
+        .canonical_body()
+        .map_err(|_| std::io::Error::other("RPC mutation body length cannot be represented"))?;
+    set_tonic_canonical_body_digest(request, &canonical_body)
+}
+
 pub fn verify_tonic_canonical_body_digest<T>(request: &tonic::Request<T>, canonical_body: &[u8]) -> std::io::Result<()> {
     let version = request
         .metadata()
@@ -466,7 +476,7 @@ pub fn verify_tonic_canonical_body_digest<T>(request: &tonic::Request<T>, canoni
     Ok(())
 }
 
-/// Verify a mutating disk RPC's canonical body digest with a rolling-upgrade fallback.
+/// Verify a mutating RPC's canonical body digest with a rolling-upgrade fallback.
 ///
 /// When the request carries a real (non-`UNSIGNED-PAYLOAD`) content SHA-256 it is verified exactly
 /// like [`verify_tonic_canonical_body_digest`]. The digest value is a member of the signed v2
@@ -497,7 +507,7 @@ fn verify_tonic_mutation_body_digest_with_strictness<T>(
         Some(digest) if digest != UNSIGNED_PAYLOAD => verify_tonic_canonical_body_digest(request, canonical_body),
         _ => {
             // RUSTFS_COMPAT_TODO(disk-mutation-body-digest): accept digestless peers during rolling upgrades. Remove after the
-            // minimum supported RustFS peer version body-binds every mutating disk RPC.
+            // minimum supported RustFS peer version body-binds every mutating RPC.
             if strict {
                 return Err(std::io::Error::other("RPC mutation requires a body-bound v2 signature"));
             }
@@ -677,11 +687,28 @@ mod tests {
     use crate::cluster::rpc::context_propagation::REQUEST_ID_HEADER;
     use crate::runtime::sources as runtime_sources;
     use http::{HeaderMap, Method};
+    use rustfs_protos::{
+        CanonicalMutationBody as _, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS,
+        proto_gen::node_service::{Mss, SignalServiceRequest},
+    };
+    use std::collections::HashMap;
     use std::io::{self, Write};
     use std::sync::{Arc, Mutex};
     use time::OffsetDateTime;
     use tracing_subscriber::fmt::MakeWriter;
 
+    fn signal_service_request(signal: &str, sub_system: &str, dry_run: &str) -> SignalServiceRequest {
+        SignalServiceRequest {
+            vars: Some(Mss {
+                value: HashMap::from([
+                    (PEER_RESTSIGNAL.to_string(), signal.to_string()),
+                    (PEER_RESTSUB_SYS.to_string(), sub_system.to_string()),
+                    (PEER_RESTDRY_RUN.to_string(), dry_run.to_string()),
+                ]),
+            }),
+        }
+    }
+
     #[derive(Clone, Default)]
     struct CapturedLogs {
         buffer: Arc<Mutex<Vec<u8>>>,
@@ -1596,6 +1623,72 @@ mod tests {
         assert_eq!(stripped.to_string(), "RPC content SHA-256 mismatch");
     }
 
+    #[test]
+    fn signal_service_mutation_contract_rejects_tampering_and_replay() {
+        ensure_test_rpc_secret();
+        let body = signal_service_request("2", "scanner", "false")
+            .canonical_body()
+            .expect("small signal request should encode");
+        let mut request = tonic::Request::new(());
+        set_tonic_canonical_body_digest(&mut request, &body).expect("canonical body digest should be attached");
+        let content_sha256 = request
+            .metadata()
+            .get(RPC_CONTENT_SHA256_HEADER)
+            .and_then(|value| value.to_str().ok());
+        let headers = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "SignalService", content_sha256)
+            .expect("body-bound auth headers should build");
+        request.metadata_mut().as_mut().extend(headers.clone());
+
+        assert!(
+            verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers).is_ok(),
+            "the first body-bound signal request must authenticate"
+        );
+        assert!(verify_tonic_mutation_body_digest(&request, &body).is_ok());
+
+        let tampered = signal_service_request("1", "scanner", "false")
+            .canonical_body()
+            .expect("small signal request should encode");
+        let error = verify_tonic_mutation_body_digest(&request, &tampered)
+            .expect_err("changing the signal must invalidate the signed digest");
+        assert_eq!(error.to_string(), "RPC content SHA-256 mismatch");
+
+        let replay = verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers)
+            .expect_err("reusing the signal nonce must fail");
+        assert_eq!(replay.to_string(), "RPC request replay detected");
+    }
+
+    #[test]
+    #[serial_test::serial(rpc_body_digest_fallback_counter)]
+    fn signal_service_mutation_contract_preserves_rollout_fallback_and_strictness() {
+        let body = signal_service_request("2", "scanner", "false")
+            .canonical_body()
+            .expect("small signal request should encode");
+        let before = global_internode_metrics().snapshot().body_digest_fallback_total;
+        let digestless = tonic::Request::new(());
+
+        assert!(
+            verify_tonic_mutation_body_digest_with_strictness(&digestless, &body, false).is_ok(),
+            "old peers must remain compatible while the rollout gate is open"
+        );
+        assert_eq!(
+            global_internode_metrics().snapshot().body_digest_fallback_total,
+            before + 1,
+            "accepted digestless signal requests must be visible in the fallback metric"
+        );
+
+        let error = verify_tonic_mutation_body_digest_with_strictness(&digestless, &body, true)
+            .expect_err("strict mode must reject a digestless signal request");
+        assert_eq!(error.to_string(), "RPC mutation requires a body-bound v2 signature");
+
+        let mut bound = tonic::Request::new(());
+        set_tonic_canonical_body_digest(&mut bound, &body).expect("canonical body digest should be attached");
+        bound
+            .metadata_mut()
+            .as_mut()
+            .insert(RPC_AUTH_VERSION_HEADER, HeaderValue::from_static(RPC_AUTH_VERSION_V2));
+        assert!(verify_tonic_mutation_body_digest_with_strictness(&bound, &body, true).is_ok());
+    }
+
     #[test]
     fn nonce_cache_rejects_replay_after_wall_clock_regression() {
         let now = Instant::now();

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

@@ -30,9 +30,9 @@ pub use client::{
 };
 pub use http_auth::{
     TONIC_RPC_PREFIX, build_auth_headers, gen_signature_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience,
-    set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, verify_ns_scanner_capability,
-    verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof,
-    verify_tonic_rpc_signature,
+    set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof,
+    verify_ns_scanner_capability, verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
+    verify_tonic_rpc_response_proof, verify_tonic_rpc_signature,
 };
 #[cfg(test)]
 pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport;

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

@@ -16,7 +16,7 @@ use crate::cluster::rpc::client::{
     TonicInterceptor, embedded_tonic_status, gen_tonic_signature_interceptor, heal_control_time_out_client,
     is_network_like_status, message_has_network_needle, node_service_time_out_client, tier_mutation_control_time_out_client,
 };
-use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof};
+use crate::cluster::rpc::{set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, verify_tonic_rpc_response_proof};
 use crate::error::{Error, Result};
 use crate::storage_api_contracts::internode::{
     SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION,
@@ -50,6 +50,7 @@ use rustfs_protos::proto_gen::node_service::{
     TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
     tier_mutation_control_service_client::TierMutationControlServiceClient,
 };
+pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS};
 use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection};
 use rustfs_utils::XHost;
 use serde::{Deserialize, Serialize as _};
@@ -69,9 +70,6 @@ use tonic::transport::Channel;
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
-pub const PEER_RESTSIGNAL: &str = "signal";
-pub const PEER_RESTSUB_SYS: &str = "sub-sys";
-pub const PEER_RESTDRY_RUN: &str = "dry-run";
 pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1;
 pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2;
 const BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024;
@@ -1160,10 +1158,11 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadBucketMetadataRequest {
+                let mut request = Request::new(LoadBucketMetadataRequest {
                     bucket: bucket.to_string(),
                     scanner_maintenance_change,
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_bucket_metadata(request).await?.into_inner();
                 if !response.success {
@@ -1183,9 +1182,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(DeleteBucketMetadataRequest {
+                let mut request = Request::new(DeleteBucketMetadataRequest {
                     bucket: bucket.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.delete_bucket_metadata(request).await?.into_inner();
                 if !response.success {
@@ -1205,9 +1205,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(DeletePolicyRequest {
+                let mut request = Request::new(DeletePolicyRequest {
                     policy_name: policy.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.delete_policy(request).await?.into_inner();
                 if !response.success {
@@ -1227,9 +1228,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadPolicyRequest {
+                let mut request = Request::new(LoadPolicyRequest {
                     policy_name: policy.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_policy(request).await?.into_inner();
                 if !response.success {
@@ -1249,11 +1251,12 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadPolicyMappingRequest {
+                let mut request = Request::new(LoadPolicyMappingRequest {
                     user_or_group: user_or_group.to_string(),
                     user_type,
                     is_group,
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_policy_mapping(request).await?.into_inner();
                 if !response.success {
@@ -1273,9 +1276,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(DeleteUserRequest {
+                let mut request = Request::new(DeleteUserRequest {
                     access_key: access_key.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.delete_user(request).await?.into_inner();
                 if !response.success {
@@ -1295,9 +1299,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(DeleteServiceAccountRequest {
+                let mut request = Request::new(DeleteServiceAccountRequest {
                     access_key: access_key.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.delete_service_account(request).await?.into_inner();
                 if !response.success {
@@ -1317,10 +1322,11 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadUserRequest {
+                let mut request = Request::new(LoadUserRequest {
                     access_key: access_key.to_string(),
                     temp,
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_user(request).await?.into_inner();
                 if !response.success {
@@ -1340,9 +1346,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadServiceAccountRequest {
+                let mut request = Request::new(LoadServiceAccountRequest {
                     access_key: access_key.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_service_account(request).await?.into_inner();
                 if !response.success {
@@ -1362,9 +1369,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadGroupRequest {
+                let mut request = Request::new(LoadGroupRequest {
                     group: group.to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_group(request).await?.into_inner();
                 if !response.success {
@@ -1384,7 +1392,8 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(ReloadSiteReplicationConfigRequest {});
+                let mut request = Request::new(ReloadSiteReplicationConfigRequest {});
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.reload_site_replication_config(request).await?.into_inner();
                 if !response.success {
@@ -1408,9 +1417,10 @@ impl PeerRestClient {
                 vars.insert(PEER_RESTSIGNAL.to_string(), sig.to_string());
                 vars.insert(PEER_RESTSUB_SYS.to_string(), sub_sys.to_string());
                 vars.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string());
-                let request = Request::new(SignalServiceRequest {
+                let mut request = Request::new(SignalServiceRequest {
                     vars: Some(Mss { value: vars }),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.signal_service(request).await?.into_inner();
                 if !response.success {
@@ -1479,7 +1489,8 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(ReloadPoolMetaRequest {});
+                let mut request = Request::new(ReloadPoolMetaRequest {});
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.reload_pool_meta(request).await?.into_inner();
                 if !response.success {
@@ -1500,9 +1511,10 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(StopRebalanceRequest {
+                let mut request = Request::new(StopRebalanceRequest {
                     expected_rebalance_id: expected_rebalance_id.unwrap_or_default().to_string(),
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.stop_rebalance(request).await?.into_inner();
                 if !response.success {
@@ -1523,7 +1535,8 @@ impl PeerRestClient {
         self.finalize_result(
             async {
                 let mut client = self.get_client().await?;
-                let request = Request::new(LoadRebalanceMetaRequest { start_rebalance });
+                let mut request = Request::new(LoadRebalanceMetaRequest { start_rebalance });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.load_rebalance_meta(request).await?.into_inner();
 
@@ -1562,7 +1575,8 @@ impl PeerRestClient {
                     })
                     .collect::<Result<Vec<_>>>()?;
                 let mut client = self.get_client().await?;
-                let request = Request::new(StartDecommissionRequest { pool_indices });
+                let mut request = Request::new(StartDecommissionRequest { pool_indices });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.start_decommission(request).await?.into_inner();
                 if !response.success {
@@ -1585,7 +1599,8 @@ impl PeerRestClient {
                 let pool_index = u32::try_from(pool_index)
                     .map_err(|_| Error::other(format!("decommission pool index {pool_index} exceeds RPC range")))?;
                 let mut client = self.get_client().await?;
-                let request = Request::new(CancelDecommissionRequest { pool_index });
+                let mut request = Request::new(CancelDecommissionRequest { pool_index });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.cancel_decommission(request).await?.into_inner();
                 if !response.success {
@@ -1608,7 +1623,8 @@ impl PeerRestClient {
                 let pool_index = u32::try_from(pool_index)
                     .map_err(|_| Error::other(format!("decommission pool index {pool_index} exceeds RPC range")))?;
                 let mut client = self.get_client().await?;
-                let request = Request::new(ClearDecommissionRequest { pool_index });
+                let mut request = Request::new(ClearDecommissionRequest { pool_index });
+                set_tonic_mutation_body_digest(&mut request)?;
 
                 let response = client.clear_decommission(request).await?.into_inner();
                 if !response.success {
@@ -1661,6 +1677,9 @@ impl PeerRestClient {
             Err(err) => return tier_config_reload_connection_outcome(err),
         };
         let mut request = Request::new(LoadTransitionTierConfigRequest {});
+        if let Err(err) = set_tonic_mutation_body_digest(&mut request) {
+            return TierConfigReloadOutcome::Terminal(Error::other(err));
+        }
         request.set_timeout(rustfs_protos::heal_control_execution_timeout());
 
         let response = match client.load_transition_tier_config(request).await {

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

@@ -16,6 +16,7 @@ use crate::bucket::metadata_sys;
 use crate::cluster::rpc::client::{
     TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client,
 };
+use crate::cluster::rpc::set_tonic_mutation_body_digest;
 use crate::disk::error::DiskError;
 use crate::disk::error::{Error, Result};
 use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
@@ -930,10 +931,11 @@ impl PeerS3Client for RemotePeerS3Client {
             || async {
                 let options: String = serde_json::to_string(opts)?;
                 let mut client = self.get_client().await?;
-                let request = Request::new(HealBucketRequest {
+                let mut request = Request::new(HealBucketRequest {
                     bucket: bucket.to_string(),
                     options,
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
                 let response = client.heal_bucket(request).await?.into_inner();
                 if !response.success {
                     return if let Some(err) = response.error {
@@ -986,10 +988,11 @@ impl PeerS3Client for RemotePeerS3Client {
             || async {
                 let options = serde_json::to_string(opts)?;
                 let mut client = self.get_client().await?;
-                let request = Request::new(MakeBucketRequest {
+                let mut request = Request::new(MakeBucketRequest {
                     name: bucket.to_string(),
                     options,
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
                 let response = client.make_bucket(request).await?.into_inner();
 
                 if !response.success {
@@ -1040,10 +1043,11 @@ impl PeerS3Client for RemotePeerS3Client {
                 let options = serde_json::to_string(opts)?;
                 let mut client = self.get_client().await?;
 
-                let request = Request::new(DeleteBucketRequest {
+                let mut request = Request::new(DeleteBucketRequest {
                     bucket: bucket.to_string(),
                     options,
                 });
+                set_tonic_mutation_body_digest(&mut request)?;
                 let response = client.delete_bucket(request).await?.into_inner();
                 if !response.success {
                     return if let Some(err) = response.error {

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use crate::cluster::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client};
+use crate::cluster::rpc::set_tonic_mutation_body_digest;
 use async_trait::async_trait;
 use bytes::Bytes;
 use rustfs_lock::{
@@ -313,10 +314,11 @@ impl LockClient for RemoteClient {
         info!("remote acquire_exclusive for {}", request.resource);
         let mut client = self.get_client().await?;
         let resource_summary = request.resource.to_string();
-        let req = Request::new(GenerallyLockRequest {
+        let mut req = Request::new(GenerallyLockRequest {
             args: serde_json::to_string(&request)
                 .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
         });
+        set_tonic_mutation_body_digest(&mut req)?;
 
         let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await {
             Ok(resp) => resp.into_inner(),
@@ -347,7 +349,7 @@ impl LockClient for RemoteClient {
 
         let mut client = self.get_client().await?;
         let resource_summary = Self::summarize_resources(requests);
-        let req = Request::new(BatchGenerallyLockRequest {
+        let mut req = Request::new(BatchGenerallyLockRequest {
             args: requests
                 .iter()
                 .map(|request| {
@@ -355,6 +357,7 @@ impl LockClient for RemoteClient {
                 })
                 .collect::<Result<Vec<_>>>()?,
         });
+        set_tonic_mutation_body_digest(&mut req)?;
 
         let resp = match self
             .execute_rpc("lock_batch", &resource_summary, client.lock_batch(req))
@@ -395,7 +398,8 @@ impl LockClient for RemoteClient {
             .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?;
         let mut client = self.get_client().await?;
         let resource_summary = unlock_request.resource.to_string();
-        let req = Request::new(GenerallyLockRequest { args: request_string });
+        let mut req = Request::new(GenerallyLockRequest { args: request_string });
+        set_tonic_mutation_body_digest(&mut req)?;
         let resp = self
             .execute_rpc("release", &resource_summary, client.un_lock(req))
             .await?
@@ -414,7 +418,7 @@ impl LockClient for RemoteClient {
         let unlock_requests = lock_ids.iter().map(Self::create_unlock_request).collect::<Vec<_>>();
         let mut client = self.get_client().await?;
         let resource_summary = Self::summarize_resources(&unlock_requests);
-        let req = Request::new(BatchGenerallyLockRequest {
+        let mut req = Request::new(BatchGenerallyLockRequest {
             args: unlock_requests
                 .iter()
                 .map(|request| {
@@ -422,6 +426,7 @@ impl LockClient for RemoteClient {
                 })
                 .collect::<Result<Vec<_>>>()?,
         });
+        set_tonic_mutation_body_digest(&mut req)?;
 
         let resp = self
             .execute_rpc("release_batch", &resource_summary, client.un_lock_batch(req))
@@ -440,10 +445,11 @@ impl LockClient for RemoteClient {
         let refresh_request = Self::create_unlock_request(lock_id);
         let mut client = self.get_client().await?;
         let resource_summary = refresh_request.resource.to_string();
-        let req = Request::new(GenerallyLockRequest {
+        let mut req = Request::new(GenerallyLockRequest {
             args: serde_json::to_string(&refresh_request)
                 .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
         });
+        set_tonic_mutation_body_digest(&mut req)?;
         let resp = self
             .execute_rpc("refresh", &resource_summary, client.refresh(req))
             .await?
@@ -459,10 +465,11 @@ impl LockClient for RemoteClient {
         let force_request = Self::create_unlock_request(lock_id);
         let mut client = self.get_client().await?;
         let resource_summary = force_request.resource.to_string();
-        let req = Request::new(GenerallyLockRequest {
+        let mut req = Request::new(GenerallyLockRequest {
             args: serde_json::to_string(&force_request)
                 .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
         });
+        set_tonic_mutation_body_digest(&mut req)?;
         let resp = self
             .execute_rpc("force_release", &resource_summary, client.force_un_lock(req))
             .await?
@@ -483,10 +490,11 @@ impl LockClient for RemoteClient {
         let mut client = self.get_client().await?;
 
         // Try to acquire a very short-lived lock to test availability
-        let req = Request::new(GenerallyLockRequest {
+        let mut req = Request::new(GenerallyLockRequest {
             args: serde_json::to_string(&status_request)
                 .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
         });
+        set_tonic_mutation_body_digest(&mut req)?;
 
         // Try exclusive lock first with very short timeout
         let resp = match self.execute_rpc("check_status", &resource_summary, client.lock(req)).await {
@@ -497,10 +505,11 @@ impl LockClient for RemoteClient {
         if resp.success {
             // If we successfully acquired the lock, the resource was free.
             // Immediately release it on a best-effort basis.
-            let release_req = Request::new(GenerallyLockRequest {
+            let mut release_req = Request::new(GenerallyLockRequest {
                 args: serde_json::to_string(&status_request)
                     .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
             });
+            set_tonic_mutation_body_digest(&mut release_req)?;
             let _ = self
                 .execute_rpc("check_status_release", &resource_summary, client.un_lock(release_req))
                 .await;

crates/heal/Cargo.toml

@@ -26,9 +26,6 @@ documentation = "https://docs.rs/rustfs-heal/latest/rustfs_heal/"
 keywords = ["RustFS", "heal", "erasure-coding", "Minio"]
 categories = ["web-programming", "development-tools", "filesystem"]
 
-[lints]
-workspace = true
-
 [dependencies]
 rustfs-config = { workspace = true }
 rustfs-concurrency = { workspace = true }
@@ -56,7 +53,7 @@ serial_test = { workspace = true }
 tempfile = { workspace = true }
 walkdir = { workspace = true }
 http = { workspace = true }
-temp-env = { workspace = true, features = ["async_closure"] }
+temp-env = { workspace = true }
 tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
 
 [lib]

crates/heal/src/heal/channel.rs

@@ -482,7 +482,7 @@ impl HealChannelProcessor {
                     request_id: client_token,
                     success: false,
                     data: None,
-                    error: Some(error_text),
+                    error: Some(error_text.clone()),
                 };
                 let _ = response_tx.send(Ok(response.clone()));
                 self.publish_response(response);

crates/heal/src/heal/event.rs

@@ -576,7 +576,7 @@ mod tests {
         assert_eq!(handler.event_count(), 1);
 
         handler.add_event(event.clone());
-        handler.add_event(event);
+        handler.add_event(event.clone());
         assert_eq!(handler.event_count(), 3);
     }
 
@@ -593,7 +593,7 @@ mod tests {
 
         handler.add_event(event.clone());
         handler.add_event(event.clone());
-        handler.add_event(event); // Should remove oldest
+        handler.add_event(event.clone()); // Should remove oldest
 
         assert_eq!(handler.event_count(), 2);
     }
@@ -610,7 +610,7 @@ mod tests {
         };
 
         handler.add_event(event.clone());
-        handler.add_event(event);
+        handler.add_event(event.clone());
 
         let events = handler.get_events();
         assert_eq!(events.len(), 2);

crates/heal/src/heal/manager.rs

@@ -2892,7 +2892,7 @@ fn update_task_running_metric_for_task(active_heals: &HashMap<String, Arc<HealTa
     gauge!(
         "rustfs_heal_task_running",
         "type" => type_label.to_string(),
-        "set" => set_label
+        "set" => set_label.clone()
     )
     .set(count as f64);
 }
@@ -3482,7 +3482,7 @@ mod tests {
         );
 
         assert_eq!(queue.push(blocked), QueuePushOutcome::Accepted);
-        assert_eq!(queue.push(runnable), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(runnable.clone()), QueuePushOutcome::Accepted);
 
         let mut running = HashMap::new();
         running.insert("pool_0_set_1".to_string(), 1);

crates/heal/tests/heal_b920_subquorum_union_test.rs

@@ -28,7 +28,6 @@ use rustfs_heal::heal::storage::{
 };
 use serial_test::serial;
 use std::{
-    future::Future,
     path::{Path, PathBuf},
     sync::Arc,
 };
@@ -53,9 +52,15 @@ fn versioned_test_data(seed: u8) -> Vec<u8> {
         .collect()
 }
 
-/// Disable the grace window while the destructive heal decision is evaluated.
-async fn with_dangling_grace_disabled<T>(future: impl Future<Output = T>) -> T {
-    temp_env::async_with_vars([(GRACE_ENV, Some("0"))], future).await
+/// Disable the dangling-delete grace window so the destructive path is genuinely
+/// LIVE in these tests: without the decision-1 guard, a recoverable version WOULD
+/// be dangling-deleted here. With grace at its 1h default the delete path would be
+/// masked and the test would prove nothing.
+fn disable_dangling_grace() {
+    // Safe under nextest: each test runs in its own process and is `#[serial]`.
+    unsafe {
+        std::env::set_var(GRACE_ENV, "0");
+    }
 }
 
 /// Build a real N-disk single-set `ECStore` + `ECStoreHealStorage` via the
@@ -245,6 +250,7 @@ mod serial_tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]
     async fn union_meta_lost_data_present_is_repaired_not_destroyed() {
+        disable_dangling_grace();
         let (disk_paths, ecstore, heal_storage) = heal_env_n(8).await;
         let bucket = "b920-meta-lost";
         let object = "obj.bin";
@@ -262,10 +268,10 @@ mod serial_tests {
         }
 
         // Heal the version through the real heal storage (Deep).
-        let (_result, error) =
-            with_dangling_grace_disabled(heal_storage.heal_object(bucket, object, Some(&v1), &deep_heal_opts()))
-                .await
-                .expect("heal_object call must not itself error");
+        let (_result, error) = heal_storage
+            .heal_object(bucket, object, Some(&v1), &deep_heal_opts())
+            .await
+            .expect("heal_object call must not itself error");
         assert!(
             error.is_none(),
             "recoverable version must heal without error (must NOT be dangling-deleted): {error:?}"
@@ -294,6 +300,7 @@ mod serial_tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]
     async fn deep_heal_torn_minority_is_dangling_deleted_with_grace_zero() {
+        disable_dangling_grace();
         let (disk_paths, ecstore, heal_storage) = heal_env_n(4).await;
         let bucket = "b920-torn";
         let object = "obj.bin";
@@ -311,10 +318,10 @@ mod serial_tests {
         // is a real destructive action, not a no-op).
         assert_eq!(count_part_files(&object_dir(&disk_paths[0], bucket, object)), 1);
 
-        let (_result, error) =
-            with_dangling_grace_disabled(heal_storage.heal_object(bucket, object, Some(&v1), &deep_heal_opts()))
-                .await
-                .expect("heal_object call must not itself error");
+        let (_result, error) = heal_storage
+            .heal_object(bucket, object, Some(&v1), &deep_heal_opts())
+            .await
+            .expect("heal_object call must not itself error");
         // A dangling delete reports the version as gone (FileVersionNotFound),
         // proving the destructive path fired for a genuinely torn write.
         assert!(error.is_some(), "a torn (< data_blocks) version must NOT be silently treated as healed");
@@ -342,6 +349,7 @@ mod serial_tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]
     async fn deep_heal_restores_subquorum_but_reconstructable_version_wider_set() {
+        disable_dangling_grace();
         let (disk_paths, ecstore, heal_storage) = heal_env_n(8).await;
         let bucket = "b920-reconstruct";
         let object = "obj.bin";
@@ -365,10 +373,10 @@ mod serial_tests {
             "disk-walk must enumerate the reconstructable sub-quorum version"
         );
 
-        let (_result, error) =
-            with_dangling_grace_disabled(heal_storage.heal_object(bucket, object, Some(&v1), &deep_heal_opts()))
-                .await
-                .expect("heal_object call must not itself error");
+        let (_result, error) = heal_storage
+            .heal_object(bucket, object, Some(&v1), &deep_heal_opts())
+            .await
+            .expect("heal_object call must not itself error");
         assert!(error.is_none(), "reconstructable version must heal cleanly: {error:?}");
 
         // part.* + xl.meta physically restored on the 4 wiped disks.
@@ -455,6 +463,7 @@ mod serial_tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]
     async fn offline_disk_during_walk_does_not_dangling_delete() {
+        disable_dangling_grace();
         let (disk_paths, ecstore, heal_storage) = heal_env_n(4).await;
         let bucket = "b920-offline";
         let object = "obj.bin";
@@ -482,7 +491,8 @@ mod serial_tests {
             remove: false,
             ..Default::default()
         };
-        let (_result, error) = with_dangling_grace_disabled(heal_storage.heal_object(bucket, object, Some(&v1), &normal_opts))
+        let (_result, error) = heal_storage
+            .heal_object(bucket, object, Some(&v1), &normal_opts)
             .await
             .expect("heal_object call must not itself error");
         assert!(error.is_none(), "quorum-present object must not be destroyed: {error:?}");
@@ -500,6 +510,7 @@ mod serial_tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]
     async fn deep_heal_keeps_present_ec2_plus_2_shards_healthy() {
+        disable_dangling_grace();
         let (disk_paths, ecstore, heal_storage) = heal_env_n(4).await;
         let bucket = "b1044-deep-verify";
         let object = "obj.bin";
@@ -513,10 +524,10 @@ mod serial_tests {
             assert_eq!(count_part_files(&object_dir(disk, bucket, object)), 1, "intact shard must remain present");
         }
 
-        let (_result, error) =
-            with_dangling_grace_disabled(heal_storage.heal_object(bucket, object, Some(&v1), &deep_heal_opts()))
-                .await
-                .expect("deep heal_object call must not itself error");
+        let (_result, error) = heal_storage
+            .heal_object(bucket, object, Some(&v1), &deep_heal_opts())
+            .await
+            .expect("deep heal_object call must not itself error");
         assert!(error.is_none(), "Deep heal must retain the three intact EC2+2 shards: {error:?}");
 
         assert!(

crates/heal/tests/heal_integration_test.rs

@@ -390,8 +390,10 @@ mod serial_tests {
 
         // ─── 1️⃣ delete format.json on one disk ──────────────
         let format_path = disk_paths[0].join(".rustfs.sys").join("format.json");
-        std::fs::remove_dir_all(&disk_paths[0]).expect("failed to delete all contents under disk_paths[0]");
+        let failed_disk_path = disk_paths[0].with_extension("failed");
+        std::fs::rename(&disk_paths[0], &failed_disk_path).expect("failed to detach disk_paths[0]");
         std::fs::create_dir_all(&disk_paths[0]).expect("failed to recreate disk_paths[0] directory");
+        assert!(!format_path.exists(), "replacement disk already contains format.json before heal");
         println!("✅ Deleted format.json on disk: {:?}", disk_paths[0]);
 
         let (_format_result, format_error) = heal_storage.heal_format(false).await.expect("failed to run heal_format");

crates/protos/src/lib.rs

@@ -447,10 +447,19 @@ impl CanonicalBodyBuilder {
         self.push_bytes(field.as_bytes())
     }
 
+    fn push_optional_str(&mut self, field: Option<&str>) -> Result<(), std::num::TryFromIntError> {
+        self.body.push(u8::from(field.is_some()));
+        self.push_str(field.unwrap_or_default())
+    }
+
     fn push_bool(&mut self, field: bool) {
         self.body.push(u8::from(field));
     }
 
+    fn push_u32(&mut self, field: u32) {
+        self.body.extend_from_slice(&field.to_be_bytes());
+    }
+
     fn push_u64(&mut self, field: u64) {
         self.body.extend_from_slice(&field.to_be_bytes());
     }
@@ -465,6 +474,211 @@ impl CanonicalBodyBuilder {
     }
 }
 
+pub const PEER_RESTSIGNAL: &str = "signal";
+pub const PEER_RESTSUB_SYS: &str = "sub-sys";
+pub const PEER_RESTDRY_RUN: &str = "dry-run";
+
+/// A stable semantic body for a side-effecting unary RPC.
+///
+/// Implementations deliberately enumerate handler-consumed fields instead of re-encoding the
+/// protobuf message, whose unknown fields and map order are not a mixed-version contract.
+pub trait CanonicalMutationBody {
+    fn canonical_body(&self) -> Result<Vec<u8>, std::num::TryFromIntError>;
+}
+
+macro_rules! impl_canonical_mutation_body {
+    ($request:ty, $domain:expr, |$value:ident, $body:ident| $fields:block) => {
+        impl CanonicalMutationBody for $request {
+            fn canonical_body(&self) -> Result<Vec<u8>, std::num::TryFromIntError> {
+                let $value = self;
+                let mut $body = CanonicalBodyBuilder::new($domain);
+                $fields
+                Ok($body.finish())
+            }
+        }
+    };
+    ($request:ty, $domain:expr) => {
+        impl CanonicalMutationBody for $request {
+            fn canonical_body(&self) -> Result<Vec<u8>, std::num::TryFromIntError> {
+                Ok(CanonicalBodyBuilder::new($domain).finish())
+            }
+        }
+    };
+}
+
+impl_canonical_mutation_body!(
+    proto_gen::node_service::SignalServiceRequest,
+    b"rustfs-signal-service-request-v1\0",
+    |request, body| {
+        let vars = request.vars.as_ref().map(|vars| &vars.value);
+        body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTSIGNAL).map(String::as_str)))?;
+        body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTSUB_SYS).map(String::as_str)))?;
+        body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTDRY_RUN).map(String::as_str)))?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::HealBucketRequest,
+    b"rustfs-heal-bucket-request-v1\0",
+    |request, body| {
+        body.push_str(&request.bucket)?;
+        body.push_str(&request.options)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::MakeBucketRequest,
+    b"rustfs-make-bucket-request-v1\0",
+    |request, body| {
+        body.push_str(&request.name)?;
+        body.push_str(&request.options)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::DeleteBucketRequest,
+    b"rustfs-delete-bucket-request-v1\0",
+    |request, body| {
+        body.push_str(&request.bucket)?;
+        body.push_str(&request.options)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::GenerallyLockRequest,
+    b"rustfs-lock-request-v1\0",
+    |request, body| {
+        body.push_str(&request.args)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::BatchGenerallyLockRequest,
+    b"rustfs-lock-batch-request-v1\0",
+    |request, body| {
+        body.push_count(request.args.len())?;
+        for arg in &request.args {
+            body.push_str(arg)?;
+        }
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadBucketMetadataRequest,
+    b"rustfs-load-bucket-metadata-request-v1\0",
+    |request, body| {
+        body.push_str(&request.bucket)?;
+        body.push_bool(request.scanner_maintenance_change);
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::DeleteBucketMetadataRequest,
+    b"rustfs-delete-bucket-metadata-request-v1\0",
+    |request, body| {
+        body.push_str(&request.bucket)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::DeletePolicyRequest,
+    b"rustfs-delete-policy-request-v1\0",
+    |request, body| {
+        body.push_str(&request.policy_name)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadPolicyRequest,
+    b"rustfs-load-policy-request-v1\0",
+    |request, body| {
+        body.push_str(&request.policy_name)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadPolicyMappingRequest,
+    b"rustfs-load-policy-mapping-request-v1\0",
+    |request, body| {
+        body.push_str(&request.user_or_group)?;
+        body.push_u64(request.user_type);
+        body.push_bool(request.is_group);
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::DeleteUserRequest,
+    b"rustfs-delete-user-request-v1\0",
+    |request, body| {
+        body.push_str(&request.access_key)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::DeleteServiceAccountRequest,
+    b"rustfs-delete-service-account-request-v1\0",
+    |request, body| {
+        body.push_str(&request.access_key)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadUserRequest,
+    b"rustfs-load-user-request-v1\0",
+    |request, body| {
+        body.push_str(&request.access_key)?;
+        body.push_bool(request.temp);
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadServiceAccountRequest,
+    b"rustfs-load-service-account-request-v1\0",
+    |request, body| {
+        body.push_str(&request.access_key)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadGroupRequest,
+    b"rustfs-load-group-request-v1\0",
+    |request, body| {
+        body.push_str(&request.group)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::ReloadSiteReplicationConfigRequest,
+    b"rustfs-reload-site-replication-config-request-v1\0"
+);
+impl_canonical_mutation_body!(proto_gen::node_service::ReloadPoolMetaRequest, b"rustfs-reload-pool-meta-request-v1\0");
+impl_canonical_mutation_body!(
+    proto_gen::node_service::StopRebalanceRequest,
+    b"rustfs-stop-rebalance-request-v1\0",
+    |request, body| {
+        body.push_str(&request.expected_rebalance_id)?;
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadRebalanceMetaRequest,
+    b"rustfs-load-rebalance-meta-request-v1\0",
+    |request, body| {
+        body.push_bool(request.start_rebalance);
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::StartDecommissionRequest,
+    b"rustfs-start-decommission-request-v1\0",
+    |request, body| {
+        body.push_count(request.pool_indices.len())?;
+        for pool_index in &request.pool_indices {
+            body.push_u32(*pool_index);
+        }
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::CancelDecommissionRequest,
+    b"rustfs-cancel-decommission-request-v1\0",
+    |request, body| {
+        body.push_u32(request.pool_index);
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::ClearDecommissionRequest,
+    b"rustfs-clear-decommission-request-v1\0",
+    |request, body| {
+        body.push_u32(request.pool_index);
+    }
+);
+impl_canonical_mutation_body!(
+    proto_gen::node_service::LoadTransitionTierConfigRequest,
+    b"rustfs-load-transition-tier-config-request-v1\0"
+);
+
 // Canonical request bodies for the mutating NodeService disk RPCs (backlog#1327 body-digest
 // binding). Each covers every semantic wire field — including both the msgpack `_bin` payload and
 // its JSON compatibility copy — so tampering with either encoding, or stripping `_bin` to force
@@ -1136,6 +1350,154 @@ mod disk_mutation_canonical_tests {
     }
 }
 
+#[cfg(test)]
+mod non_disk_mutation_canonical_tests {
+    use super::*;
+    use proto_gen::node_service::*;
+    use std::collections::HashMap;
+
+    macro_rules! assert_fields_bound {
+        ($request:ty, {$($field:ident: $value:expr),+ $(,)?}) => {{
+            let baseline = <$request>::default();
+            let expected = baseline.canonical_body().expect("baseline canonical body should encode");
+            $(
+                let mut variant = baseline.clone();
+                variant.$field = $value;
+                assert_ne!(
+                    expected,
+                    variant.canonical_body().expect("variant canonical body should encode"),
+                    concat!(stringify!($request), " omitted field ", stringify!($field)),
+                );
+            )+
+        }};
+    }
+
+    fn signal_request(signal: Option<&str>, sub_system: Option<&str>, dry_run: Option<&str>) -> SignalServiceRequest {
+        let mut value = HashMap::new();
+        if let Some(signal) = signal {
+            value.insert(PEER_RESTSIGNAL.to_string(), signal.to_string());
+        }
+        if let Some(sub_system) = sub_system {
+            value.insert(PEER_RESTSUB_SYS.to_string(), sub_system.to_string());
+        }
+        if let Some(dry_run) = dry_run {
+            value.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string());
+        }
+        SignalServiceRequest {
+            vars: Some(Mss { value }),
+        }
+    }
+
+    #[test]
+    fn signal_service_canonical_body_is_versioned_and_binds_every_semantic_field() {
+        let request = signal_request(Some("2"), Some("scanner"), Some("false"));
+        let baseline = request.canonical_body().expect("small signal request should encode");
+        assert!(baseline.starts_with(b"rustfs-signal-service-request-v1\0"));
+
+        for variant in [
+            signal_request(Some("1"), Some("scanner"), Some("false")),
+            signal_request(Some("2"), Some("heal"), Some("false")),
+            signal_request(Some("2"), Some("scanner"), Some("true")),
+            signal_request(None, Some("scanner"), Some("false")),
+            signal_request(Some("2"), None, Some("false")),
+            signal_request(Some("2"), Some("scanner"), None),
+        ] {
+            assert_ne!(baseline, variant.canonical_body().expect("small signal request should encode"));
+        }
+
+        let mut ignored_field = request;
+        ignored_field
+            .vars
+            .as_mut()
+            .expect("signal vars should exist")
+            .value
+            .insert("future-field".to_string(), "ignored".to_string());
+        assert_eq!(
+            baseline,
+            ignored_field
+                .canonical_body()
+                .expect("unknown signal field should not affect the semantic body"),
+        );
+    }
+
+    #[test]
+    fn signal_service_canonical_body_distinguishes_missing_and_empty_fields() {
+        assert_ne!(
+            signal_request(None, Some("scanner"), Some("false"))
+                .canonical_body()
+                .expect("missing signal should encode"),
+            signal_request(Some(""), Some("scanner"), Some("false"))
+                .canonical_body()
+                .expect("empty signal should encode"),
+        );
+    }
+
+    #[test]
+    fn bucket_and_lock_canonical_bodies_bind_every_semantic_field() {
+        assert_fields_bound!(HealBucketRequest, { bucket: "bucket".into(), options: "opts".into() });
+        assert_fields_bound!(MakeBucketRequest, { name: "bucket".into(), options: "opts".into() });
+        assert_fields_bound!(DeleteBucketRequest, { bucket: "bucket".into(), options: "opts".into() });
+        assert_fields_bound!(GenerallyLockRequest, { args: "lock".into() });
+        assert_fields_bound!(BatchGenerallyLockRequest, { args: vec!["first".into(), "second".into()] });
+
+        let first = BatchGenerallyLockRequest {
+            args: vec!["first".into(), "second".into()],
+        };
+        let reversed = BatchGenerallyLockRequest {
+            args: vec!["second".into(), "first".into()],
+        };
+        assert_ne!(first.canonical_body().unwrap(), reversed.canonical_body().unwrap());
+    }
+
+    #[test]
+    fn metadata_and_iam_canonical_bodies_bind_every_semantic_field() {
+        assert_fields_bound!(LoadBucketMetadataRequest, {
+            bucket: "bucket".into(),
+            scanner_maintenance_change: true,
+        });
+        assert_fields_bound!(DeleteBucketMetadataRequest, { bucket: "bucket".into() });
+        assert_fields_bound!(DeletePolicyRequest, { policy_name: "policy".into() });
+        assert_fields_bound!(LoadPolicyRequest, { policy_name: "policy".into() });
+        assert_fields_bound!(LoadPolicyMappingRequest, {
+            user_or_group: "user".into(),
+            user_type: 1,
+            is_group: true,
+        });
+        assert_fields_bound!(DeleteUserRequest, { access_key: "user".into() });
+        assert_fields_bound!(DeleteServiceAccountRequest, { access_key: "service".into() });
+        assert_fields_bound!(LoadUserRequest, { access_key: "user".into(), temp: true });
+        assert_fields_bound!(LoadServiceAccountRequest, { access_key: "service".into() });
+        assert_fields_bound!(LoadGroupRequest, { group: "group".into() });
+    }
+
+    #[test]
+    fn control_plane_canonical_bodies_bind_every_semantic_field() {
+        assert_fields_bound!(StopRebalanceRequest, { expected_rebalance_id: "rebalance".into() });
+        assert_fields_bound!(LoadRebalanceMetaRequest, { start_rebalance: true });
+        assert_fields_bound!(StartDecommissionRequest, { pool_indices: vec![1, 2] });
+        assert_fields_bound!(CancelDecommissionRequest, { pool_index: 1 });
+        assert_fields_bound!(ClearDecommissionRequest, { pool_index: 1 });
+
+        let first = StartDecommissionRequest {
+            pool_indices: vec![1, 2],
+        };
+        let reversed = StartDecommissionRequest {
+            pool_indices: vec![2, 1],
+        };
+        assert_ne!(first.canonical_body().unwrap(), reversed.canonical_body().unwrap());
+
+        let empty_domains = [
+            ReloadSiteReplicationConfigRequest::default().canonical_body().unwrap(),
+            ReloadPoolMetaRequest::default().canonical_body().unwrap(),
+            LoadTransitionTierConfigRequest::default().canonical_body().unwrap(),
+        ];
+        assert!(empty_domains.iter().all(|body| !body.is_empty()));
+        assert_ne!(empty_domains[0], empty_domains[1]);
+        assert_ne!(empty_domains[0], empty_domains[2]);
+        assert_ne!(empty_domains[1], empty_domains[2]);
+    }
+}
+
 #[cfg(test)]
 mod scanner_activity_tests {
     use super::{

crates/protos/src/node.proto

@@ -988,107 +988,107 @@ message GetLiveEventsResponse {
 
 service NodeService {
 /* -------------------------------meta service-------------------------- */
-  rpc Ping(PingRequest) returns (PingResponse) {};
-  rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {};
-  rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {};
-  rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {};
-  rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {};
-  rpc DeleteBucket(DeleteBucketRequest) returns (DeleteBucketResponse) {};
+  rpc Ping(PingRequest) returns (PingResponse) {}; // auth-policy: read-only
+  rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {}; // auth-policy: body-bound
+  rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {}; // auth-policy: read-only
+  rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {}; // auth-policy: body-bound
+  rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {}; // auth-policy: read-only
+  rpc DeleteBucket(DeleteBucketRequest) returns (DeleteBucketResponse) {}; // auth-policy: body-bound
 
 /* -------------------------------disk service-------------------------- */
 
-  rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {};
-  rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {};
-  rpc Delete(DeleteRequest) returns (DeleteResponse) {};
-  rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {};
-  rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {};
-  rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {};
-  rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {};
-  rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {};
-  rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {};
-  rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {};
-  rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {};
-  rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {};
-  rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {};
-  rpc Write(WriteRequest) returns (WriteResponse) {};
-  rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {};
+  rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; // auth-policy: read-only
+  rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; // auth-policy: body-bound
+  rpc Delete(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound
+  rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound
+  rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound
+  rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {}; // auth-policy: body-bound
+  rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {}; // auth-policy: read-only
+  rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {}; // auth-policy: read-only
+  rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {}; // auth-policy: read-only
+  rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {}; // auth-policy: body-bound
+  rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {}; // auth-policy: body-bound
+  rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {}; // auth-policy: body-bound
+  rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {}; // auth-policy: body-bound
+  rpc Write(WriteRequest) returns (WriteResponse) {}; // auth-policy: unimplemented
+  rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {}; // auth-policy: streaming
 //  rpc Append(AppendRequest) returns (AppendResponse) {};
-  rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {};
-  rpc ListDir(ListDirRequest) returns (ListDirResponse) {};
-  rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {};
-  rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {};
-  rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {};
-  rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {};
-  rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {};
-  rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {};
-  rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {};
-  rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {};
-  rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {};
-  rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {};
-  rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {};
-  rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {};
-  rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {};
-  rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {};
-  rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {};
-  rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {};
-  rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {};
-  rpc DiskInfo(DiskInfoRequest) returns (DiskInfoResponse) {};
+  rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {}; // auth-policy: streaming
+  rpc ListDir(ListDirRequest) returns (ListDirResponse) {}; // auth-policy: read-only
+  rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {}; // auth-policy: streaming
+  rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {}; // auth-policy: body-bound
+  rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {}; // auth-policy: body-bound
+  rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {}; // auth-policy: body-bound
+  rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {}; // auth-policy: read-only
+  rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {}; // auth-policy: read-only
+  rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {}; // auth-policy: body-bound
+  rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {}; // auth-policy: body-bound
+  rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {}; // auth-policy: read-only
+  rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {}; // auth-policy: body-bound
+  rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {}; // auth-policy: read-only
+  rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {}; // auth-policy: read-only
+  rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {}; // auth-policy: read-only
+  rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {}; // auth-policy: body-bound
+  rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {}; // auth-policy: body-bound
+  rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {}; // auth-policy: read-only
+  rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {}; // auth-policy: body-bound
+  rpc DiskInfo(DiskInfoRequest) returns (DiskInfoResponse) {}; // auth-policy: read-only
   
 
 /* -------------------------------lock service-------------------------- */
 
-  rpc Lock(GenerallyLockRequest) returns (GenerallyLockResponse) {};
-  rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {};
-  rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {};
-  rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {};
-  rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {};
-  rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {};
+  rpc Lock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
+  rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
+  rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
+  rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
+  rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; // auth-policy: body-bound
+  rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; // auth-policy: body-bound
 
 /* -------------------------------peer rest service-------------------------- */
 
-  rpc LocalStorageInfo(LocalStorageInfoRequest) returns (LocalStorageInfoResponse) {};
-  rpc ServerInfo(ServerInfoRequest) returns (ServerInfoResponse) {};
-  rpc GetCpus(GetCpusRequest) returns (GetCpusResponse) {};
-  rpc GetNetInfo(GetNetInfoRequest) returns (GetNetInfoResponse) {};
-  rpc GetPartitions(GetPartitionsRequest) returns (GetPartitionsResponse) {};
-  rpc GetOsInfo(GetOsInfoRequest) returns (GetOsInfoResponse) {};
-  rpc GetSELinuxInfo(GetSELinuxInfoRequest) returns (GetSELinuxInfoResponse) {};
-  rpc GetSysConfig(GetSysConfigRequest) returns (GetSysConfigResponse) {};
-  rpc GetSysErrors(GetSysErrorsRequest) returns (GetSysErrorsResponse) {};
-  rpc GetMemInfo(GetMemInfoRequest) returns (GetMemInfoResponse) {};
-  rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {};
-  rpc GetProcInfo(GetProcInfoRequest) returns (GetProcInfoResponse) {};
-  rpc StartProfiling(StartProfilingRequest) returns (StartProfilingResponse) {};
-  rpc DownloadProfileData(DownloadProfileDataRequest) returns (DownloadProfileDataResponse) {};
-  rpc GetBucketStats(GetBucketStatsDataRequest) returns (GetBucketStatsDataResponse) {};
-  rpc GetSRMetrics(GetSRMetricsDataRequest) returns (GetSRMetricsDataResponse) {};
-  rpc GetAllBucketStats(GetAllBucketStatsRequest) returns (GetAllBucketStatsResponse) {};
-  rpc LoadBucketMetadata(LoadBucketMetadataRequest) returns (LoadBucketMetadataResponse) {};
-  rpc DeleteBucketMetadata(DeleteBucketMetadataRequest) returns (DeleteBucketMetadataResponse) {};
-  rpc DeletePolicy(DeletePolicyRequest) returns (DeletePolicyResponse) {};
-  rpc LoadPolicy(LoadPolicyRequest) returns (LoadPolicyResponse) {};
-  rpc LoadPolicyMapping(LoadPolicyMappingRequest) returns (LoadPolicyMappingResponse) {};
-  rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse) {};
-  rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {};
-  rpc LoadUser(LoadUserRequest) returns (LoadUserResponse) {};
-  rpc LoadServiceAccount(LoadServiceAccountRequest) returns (LoadServiceAccountResponse) {};
-  rpc LoadGroup(LoadGroupRequest) returns (LoadGroupResponse) {};
-  rpc ReloadSiteReplicationConfig(ReloadSiteReplicationConfigRequest) returns (ReloadSiteReplicationConfigResponse) {};
+  rpc LocalStorageInfo(LocalStorageInfoRequest) returns (LocalStorageInfoResponse) {}; // auth-policy: read-only
+  rpc ServerInfo(ServerInfoRequest) returns (ServerInfoResponse) {}; // auth-policy: read-only
+  rpc GetCpus(GetCpusRequest) returns (GetCpusResponse) {}; // auth-policy: read-only
+  rpc GetNetInfo(GetNetInfoRequest) returns (GetNetInfoResponse) {}; // auth-policy: read-only
+  rpc GetPartitions(GetPartitionsRequest) returns (GetPartitionsResponse) {}; // auth-policy: read-only
+  rpc GetOsInfo(GetOsInfoRequest) returns (GetOsInfoResponse) {}; // auth-policy: read-only
+  rpc GetSELinuxInfo(GetSELinuxInfoRequest) returns (GetSELinuxInfoResponse) {}; // auth-policy: read-only
+  rpc GetSysConfig(GetSysConfigRequest) returns (GetSysConfigResponse) {}; // auth-policy: read-only
+  rpc GetSysErrors(GetSysErrorsRequest) returns (GetSysErrorsResponse) {}; // auth-policy: read-only
+  rpc GetMemInfo(GetMemInfoRequest) returns (GetMemInfoResponse) {}; // auth-policy: read-only
+  rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}; // auth-policy: read-only
+  rpc GetProcInfo(GetProcInfoRequest) returns (GetProcInfoResponse) {}; // auth-policy: read-only
+  rpc StartProfiling(StartProfilingRequest) returns (StartProfilingResponse) {}; // auth-policy: unimplemented
+  rpc DownloadProfileData(DownloadProfileDataRequest) returns (DownloadProfileDataResponse) {}; // auth-policy: unimplemented
+  rpc GetBucketStats(GetBucketStatsDataRequest) returns (GetBucketStatsDataResponse) {}; // auth-policy: read-only
+  rpc GetSRMetrics(GetSRMetricsDataRequest) returns (GetSRMetricsDataResponse) {}; // auth-policy: unimplemented
+  rpc GetAllBucketStats(GetAllBucketStatsRequest) returns (GetAllBucketStatsResponse) {}; // auth-policy: unimplemented
+  rpc LoadBucketMetadata(LoadBucketMetadataRequest) returns (LoadBucketMetadataResponse) {}; // auth-policy: body-bound
+  rpc DeleteBucketMetadata(DeleteBucketMetadataRequest) returns (DeleteBucketMetadataResponse) {}; // auth-policy: body-bound
+  rpc DeletePolicy(DeletePolicyRequest) returns (DeletePolicyResponse) {}; // auth-policy: body-bound
+  rpc LoadPolicy(LoadPolicyRequest) returns (LoadPolicyResponse) {}; // auth-policy: body-bound
+  rpc LoadPolicyMapping(LoadPolicyMappingRequest) returns (LoadPolicyMappingResponse) {}; // auth-policy: body-bound
+  rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse) {}; // auth-policy: body-bound
+  rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {}; // auth-policy: body-bound
+  rpc LoadUser(LoadUserRequest) returns (LoadUserResponse) {}; // auth-policy: body-bound
+  rpc LoadServiceAccount(LoadServiceAccountRequest) returns (LoadServiceAccountResponse) {}; // auth-policy: body-bound
+  rpc LoadGroup(LoadGroupRequest) returns (LoadGroupResponse) {}; // auth-policy: body-bound
+  rpc ReloadSiteReplicationConfig(ReloadSiteReplicationConfigRequest) returns (ReloadSiteReplicationConfigResponse) {}; // auth-policy: body-bound
   // rpc VerifyBinary() returns () {};
   // rpc CommitBinary() returns () {};
-  rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {};
-  rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {};
-  rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {};
-  rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {};
-  rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {};
-  rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {};
-  rpc StopRebalance(StopRebalanceRequest) returns (StopRebalanceResponse) {};
-  rpc LoadRebalanceMeta(LoadRebalanceMetaRequest) returns (LoadRebalanceMetaResponse) {};
-  rpc StartDecommission(StartDecommissionRequest) returns (StartDecommissionResponse) {};
-  rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {};
-  rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {};
-  rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {};
-  rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {};
+  rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {}; // auth-policy: body-bound
+  rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {}; // auth-policy: body-bound
+  rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {}; // auth-policy: read-only
+  rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {}; // auth-policy: unimplemented
+  rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {}; // auth-policy: unimplemented
+  rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {}; // auth-policy: body-bound
+  rpc StopRebalance(StopRebalanceRequest) returns (StopRebalanceResponse) {}; // auth-policy: body-bound
+  rpc LoadRebalanceMeta(LoadRebalanceMetaRequest) returns (LoadRebalanceMetaResponse) {}; // auth-policy: body-bound
+  rpc StartDecommission(StartDecommissionRequest) returns (StartDecommissionResponse) {}; // auth-policy: body-bound
+  rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {}; // auth-policy: body-bound
+  rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {}; // auth-policy: body-bound
+  rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {}; // auth-policy: body-bound
+  rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {}; // auth-policy: read-only
 }
 
 service HealControlService {

rustfs/src/storage/rpc/node_service.rs

@@ -28,7 +28,9 @@ use crate::storage::storage_api::rpc_consumer::node_service::{
     reload_transition_tier_config,
 };
 use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, runtime_sources};
-use crate::storage::storage_api::{sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest};
+use crate::storage::storage_api::{
+    sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
+};
 use bytes::Bytes;
 use futures::Stream;
 use futures_util::future::join_all;
@@ -40,6 +42,7 @@ use rustfs_filemeta::MetacacheReader;
 use rustfs_iam::store::UserType;
 use rustfs_lock::LockClient;
 use rustfs_protos::{
+    CanonicalMutationBody,
     models::{PingBody, PingBodyBuilder},
     proto_gen::node_service::{node_service_server::NodeService as Node, *},
 };
@@ -87,6 +90,15 @@ fn signal_service_response(success: bool, error_info: Option<String>) -> Respons
     })
 }
 
+fn verify_node_mutation_body<T: CanonicalMutationBody>(request: &Request<T>, operation: &'static str) -> Result<(), Status> {
+    let canonical_body = request
+        .get_ref()
+        .canonical_body()
+        .map_err(|_| Status::invalid_argument(format!("{operation} request length cannot be represented")))?;
+    verify_tonic_mutation_body_digest(request, &canonical_body)
+        .map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}")))
+}
+
 fn supports_dynamic_config_rpc(sub_system: &str) -> bool {
     NOTIFY_SUB_SYSTEMS.contains(&sub_system)
         || matches!(
@@ -890,6 +902,7 @@ impl Node for NodeService {
     }
 
     async fn heal_bucket(&self, request: Request<HealBucketRequest>) -> Result<Response<HealBucketResponse>, Status> {
+        verify_node_mutation_body(&request, "heal bucket")?;
         self.handle_heal_bucket(request).await
     }
 
@@ -898,6 +911,7 @@ impl Node for NodeService {
     }
 
     async fn make_bucket(&self, request: Request<MakeBucketRequest>) -> Result<Response<MakeBucketResponse>, Status> {
+        verify_node_mutation_body(&request, "make bucket")?;
         self.handle_make_bucket(request).await
     }
 
@@ -906,6 +920,7 @@ impl Node for NodeService {
     }
 
     async fn delete_bucket(&self, request: Request<DeleteBucketRequest>) -> Result<Response<DeleteBucketResponse>, Status> {
+        verify_node_mutation_body(&request, "delete bucket")?;
         self.handle_delete_bucket(request).await
     }
 
@@ -1195,18 +1210,22 @@ impl Node for NodeService {
     }
 
     async fn lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
+        verify_node_mutation_body(&request, "lock")?;
         self.handle_lock(request).await
     }
 
     async fn un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
+        verify_node_mutation_body(&request, "unlock")?;
         self.handle_un_lock(request).await
     }
 
     async fn force_un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
+        verify_node_mutation_body(&request, "force unlock")?;
         self.handle_force_un_lock(request).await
     }
 
     async fn refresh(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
+        verify_node_mutation_body(&request, "refresh lock")?;
         self.handle_refresh(request).await
     }
 
@@ -1214,6 +1233,7 @@ impl Node for NodeService {
         &self,
         request: Request<BatchGenerallyLockRequest>,
     ) -> Result<Response<BatchGenerallyLockResponse>, Status> {
+        verify_node_mutation_body(&request, "lock batch")?;
         self.handle_lock_batch(request).await
     }
 
@@ -1221,6 +1241,7 @@ impl Node for NodeService {
         &self,
         request: Request<BatchGenerallyLockRequest>,
     ) -> Result<Response<BatchGenerallyLockResponse>, Status> {
+        verify_node_mutation_body(&request, "unlock batch")?;
         self.handle_un_lock_batch(request).await
     }
 
@@ -1340,6 +1361,7 @@ impl Node for NodeService {
         &self,
         request: Request<LoadBucketMetadataRequest>,
     ) -> Result<Response<LoadBucketMetadataResponse>, Status> {
+        verify_node_mutation_body(&request, "load bucket metadata")?;
         self.handle_load_bucket_metadata(request).await
     }
 
@@ -1347,10 +1369,12 @@ impl Node for NodeService {
         &self,
         request: Request<DeleteBucketMetadataRequest>,
     ) -> Result<Response<DeleteBucketMetadataResponse>, Status> {
+        verify_node_mutation_body(&request, "delete bucket metadata")?;
         self.handle_delete_bucket_metadata(request).await
     }
 
     async fn delete_policy(&self, request: Request<DeletePolicyRequest>) -> Result<Response<DeletePolicyResponse>, Status> {
+        verify_node_mutation_body(&request, "delete policy")?;
         let request = request.into_inner();
         let policy = request.policy_name;
         if policy.is_empty() {
@@ -1381,6 +1405,7 @@ impl Node for NodeService {
     }
 
     async fn load_policy(&self, request: Request<LoadPolicyRequest>) -> Result<Response<LoadPolicyResponse>, Status> {
+        verify_node_mutation_body(&request, "load policy")?;
         let request = request.into_inner();
         let policy = request.policy_name;
         if policy.is_empty() {
@@ -1413,6 +1438,7 @@ impl Node for NodeService {
         &self,
         request: Request<LoadPolicyMappingRequest>,
     ) -> Result<Response<LoadPolicyMappingResponse>, Status> {
+        verify_node_mutation_body(&request, "load policy mapping")?;
         let request = request.into_inner();
         let user_or_group = request.user_or_group;
         if user_or_group.is_empty() {
@@ -1448,6 +1474,7 @@ impl Node for NodeService {
     }
 
     async fn delete_user(&self, request: Request<DeleteUserRequest>) -> Result<Response<DeleteUserResponse>, Status> {
+        verify_node_mutation_body(&request, "delete user")?;
         let request = request.into_inner();
         let access_key = request.access_key;
         if access_key.is_empty() {
@@ -1480,6 +1507,7 @@ impl Node for NodeService {
         &self,
         request: Request<DeleteServiceAccountRequest>,
     ) -> Result<Response<DeleteServiceAccountResponse>, Status> {
+        verify_node_mutation_body(&request, "delete service account")?;
         let request = request.into_inner();
         let access_key = request.access_key;
         if access_key.is_empty() {
@@ -1515,6 +1543,7 @@ impl Node for NodeService {
     }
 
     async fn load_user(&self, request: Request<LoadUserRequest>) -> Result<Response<LoadUserResponse>, Status> {
+        verify_node_mutation_body(&request, "load user")?;
         let request = request.into_inner();
         let access_key = request.access_key;
         let temp = request.temp;
@@ -1552,6 +1581,7 @@ impl Node for NodeService {
         &self,
         request: Request<LoadServiceAccountRequest>,
     ) -> Result<Response<LoadServiceAccountResponse>, Status> {
+        verify_node_mutation_body(&request, "load service account")?;
         let request = request.into_inner();
         let access_key = request.access_key;
         if access_key.is_empty() {
@@ -1583,6 +1613,7 @@ impl Node for NodeService {
     }
 
     async fn load_group(&self, request: Request<LoadGroupRequest>) -> Result<Response<LoadGroupResponse>, Status> {
+        verify_node_mutation_body(&request, "load group")?;
         let request = request.into_inner();
         let group = request.group;
         if group.is_empty() {
@@ -1614,8 +1645,9 @@ impl Node for NodeService {
 
     async fn reload_site_replication_config(
         &self,
-        _request: Request<ReloadSiteReplicationConfigRequest>,
+        request: Request<ReloadSiteReplicationConfigRequest>,
     ) -> Result<Response<ReloadSiteReplicationConfigResponse>, Status> {
+        verify_node_mutation_body(&request, "reload site replication config")?;
         let Some(_store) = self.resolve_object_store() else {
             return Ok(Response::new(ReloadSiteReplicationConfigResponse {
                 success: false,
@@ -1635,6 +1667,7 @@ impl Node for NodeService {
     }
 
     async fn signal_service(&self, request: Request<SignalServiceRequest>) -> Result<Response<SignalServiceResponse>, Status> {
+        verify_node_mutation_body(&request, "signal service")?;
         let request = request.into_inner();
         let vars = match request.vars {
             Some(vars) => vars.value,
@@ -1835,8 +1868,9 @@ impl Node for NodeService {
 
     async fn reload_pool_meta(
         &self,
-        _request: Request<ReloadPoolMetaRequest>,
+        request: Request<ReloadPoolMetaRequest>,
     ) -> Result<Response<ReloadPoolMetaResponse>, Status> {
+        verify_node_mutation_body(&request, "reload pool metadata")?;
         let Some(store) = self.resolve_object_store() else {
             return Ok(Response::new(ReloadPoolMetaResponse {
                 success: false,
@@ -1862,6 +1896,7 @@ impl Node for NodeService {
     }
 
     async fn stop_rebalance(&self, request: Request<StopRebalanceRequest>) -> Result<Response<StopRebalanceResponse>, Status> {
+        verify_node_mutation_body(&request, "stop rebalance")?;
         let Some(store) = self.resolve_object_store() else {
             return Ok(Response::new(StopRebalanceResponse {
                 success: false,
@@ -1882,6 +1917,7 @@ impl Node for NodeService {
         &self,
         request: Request<LoadRebalanceMetaRequest>,
     ) -> Result<Response<LoadRebalanceMetaResponse>, Status> {
+        verify_node_mutation_body(&request, "load rebalance metadata")?;
         let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner();
         let Some(store) = self.resolve_object_store() else {
             log_load_rebalance_meta_rejected!("server_not_initialized", start_rebalance);
@@ -1927,6 +1963,7 @@ impl Node for NodeService {
         &self,
         request: Request<StartDecommissionRequest>,
     ) -> Result<Response<StartDecommissionResponse>, Status> {
+        verify_node_mutation_body(&request, "start decommission")?;
         let Some(store) = runtime_sources::current_object_store_handle() else {
             return Ok(Response::new(StartDecommissionResponse {
                 success: false,
@@ -1958,6 +1995,7 @@ impl Node for NodeService {
         &self,
         request: Request<CancelDecommissionRequest>,
     ) -> Result<Response<CancelDecommissionResponse>, Status> {
+        verify_node_mutation_body(&request, "cancel decommission")?;
         let Some(store) = runtime_sources::current_object_store_handle() else {
             return Ok(Response::new(CancelDecommissionResponse {
                 success: false,
@@ -1990,6 +2028,7 @@ impl Node for NodeService {
         &self,
         request: Request<ClearDecommissionRequest>,
     ) -> Result<Response<ClearDecommissionResponse>, Status> {
+        verify_node_mutation_body(&request, "clear decommission")?;
         let Some(store) = runtime_sources::current_object_store_handle() else {
             return Ok(Response::new(ClearDecommissionResponse {
                 success: false,
@@ -2020,8 +2059,9 @@ impl Node for NodeService {
 
     async fn load_transition_tier_config(
         &self,
-        _request: Request<LoadTransitionTierConfigRequest>,
+        request: Request<LoadTransitionTierConfigRequest>,
     ) -> Result<Response<LoadTransitionTierConfigResponse>, Status> {
+        verify_node_mutation_body(&request, "load transition tier config")?;
         let Some(store) = self.resolve_object_store() else {
             return Ok(Response::new(LoadTransitionTierConfigResponse {
                 success: false,
@@ -2072,21 +2112,24 @@ mod tests {
         sys::NewServiceAccountOpts,
     };
     use rustfs_kms::KmsServiceManager;
+    use rustfs_protos::CanonicalMutationBody as _;
     use rustfs_protos::models::PingBodyBuilder;
     use rustfs_protos::proto_gen::node_service::{
-        BackgroundHealStatusRequest, CheckPartsRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest,
-        DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest,
-        DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest,
-        GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest,
-        GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest,
-        GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest,
-        HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest,
-        LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
+        BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest,
+        ClearDecommissionRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, DeletePolicyRequest,
+        DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, DeleteVersionsRequest,
+        DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, GetAllBucketStatsRequest,
+        GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest,
+        GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest,
+        GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, HealBucketRequest, HealControlRequest,
+        ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest,
+        LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
         LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest,
         MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest,
         ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest,
         RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ServerInfoRequest, SettlePartTransactionRequest,
-        SignalServiceRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState,
+        SignalServiceRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
+        StartDecommissionRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState,
         TierMutationPrepareRequest, UpdateMetacacheListingRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
         WriteMetadataRequest, WriteRequest,
         heal_control_service_client::HealControlServiceClient,
@@ -2095,13 +2138,79 @@ mod tests {
         node_service_server::NodeServiceServer,
         tier_mutation_control_service_server::TierMutationControlService as _,
     };
-    use std::{collections::HashMap, sync::Arc};
+    use std::{
+        collections::{HashMap, HashSet},
+        sync::Arc,
+    };
     use time::OffsetDateTime;
     use tokio::net::TcpListener;
     use tokio::time::Duration;
     use tokio_stream::wrappers::TcpListenerStream;
     use tonic::{Request, Response, Status};
 
+    const DISK_MUTATION_RPC_METHODS: [&str; 18] = [
+        "renamedata",
+        "deleteversion",
+        "deleteversions",
+        "writemetadata",
+        "updatemetadata",
+        "writeall",
+        "delete",
+        "deletepaths",
+        "renamefile",
+        "renamepart",
+        "prepareparttransaction",
+        "settleparttransaction",
+        "deletevolume",
+        "makevolume",
+        "makevolumes",
+        "acquiresnapshotlease",
+        "renewsnapshotlease",
+        "releasesnapshotlease",
+    ];
+
+    fn normalized_rpc_method(method: &str) -> String {
+        method.replace('_', "").to_ascii_lowercase()
+    }
+
+    fn node_service_auth_policies() -> HashMap<String, &'static str> {
+        const POLICY_MARKER: &str = "// auth-policy: ";
+
+        let schema = include_str!("../../../../crates/protos/src/node.proto");
+        let service = schema
+            .split_once("service NodeService {")
+            .expect("NodeService must exist in node.proto")
+            .1
+            .split_once("\n}")
+            .expect("NodeService must have a closing brace")
+            .0;
+        let mut policies = HashMap::new();
+        for declaration in service.lines().filter_map(|line| line.trim().strip_prefix("rpc ")) {
+            let (rpc, policy) = declaration
+                .split_once(POLICY_MARKER)
+                .expect("every NodeService RPC must declare an auth-policy beside its proto definition");
+            let method = rpc.split_once('(').expect("RPC declaration must have a request type").0;
+            assert!(
+                policies.insert(normalized_rpc_method(method), policy.trim()).is_none(),
+                "duplicate NodeService RPC {method}",
+            );
+        }
+        assert!(!policies.is_empty(), "NodeService must declare RPC methods");
+        policies
+    }
+
+    #[test]
+    fn every_node_service_rpc_declares_an_auth_policy() {
+        const VALID_POLICIES: [&str; 4] = ["body-bound", "read-only", "streaming", "unimplemented"];
+
+        for (method, policy) in node_service_auth_policies() {
+            assert!(
+                VALID_POLICIES.contains(&policy),
+                "NodeService RPC {method} has unsupported auth-policy {policy:?}",
+            );
+        }
+    }
+
     struct HealControlMockStorage;
 
     #[async_trait::async_trait]
@@ -2485,9 +2594,14 @@ mod tests {
     async fn every_mutating_handler_enforces_its_body_digest() {
         let service = make_server();
         let disk = "http://node-a:9000/data/rustfs0".to_string();
+        let mut covered_methods = HashSet::new();
 
         macro_rules! assert_gated {
             ($method:ident, $msg:expr, $canonical:path) => {{
+                assert!(
+                    covered_methods.insert(normalized_rpc_method(stringify!($method))),
+                    concat!("duplicate disk mutation test for ", stringify!($method)),
+                );
                 let msg = $msg;
 
                 // Correct digest: the gate passes and the handler proceeds to the (unknown) disk
@@ -2603,6 +2717,37 @@ mod tests {
             },
             rustfs_protos::canonical_delete_request_body
         );
+        assert_gated!(
+            acquire_snapshot_lease,
+            SnapshotLeaseRequest {
+                disk: disk.clone(),
+                volume: "v".into(),
+                path: "p".into(),
+                ttl_ms: 60_000,
+            },
+            rustfs_protos::canonical_snapshot_lease_request_body
+        );
+        assert_gated!(
+            renew_snapshot_lease,
+            SnapshotLeaseRenewRequest {
+                disk: disk.clone(),
+                volume: "v".into(),
+                path: "p".into(),
+                token: vec![1; 16].into(),
+                ttl_ms: 60_000,
+            },
+            rustfs_protos::canonical_snapshot_lease_renew_request_body
+        );
+        assert_gated!(
+            release_snapshot_lease,
+            SnapshotLeaseReleaseRequest {
+                disk: disk.clone(),
+                volume: "v".into(),
+                path: "p".into(),
+                token: vec![1; 16].into(),
+            },
+            rustfs_protos::canonical_snapshot_lease_release_request_body
+        );
         assert_gated!(
             delete_paths,
             DeletePathsRequest {
@@ -2682,6 +2827,12 @@ mod tests {
             },
             rustfs_protos::canonical_make_volumes_request_body
         );
+
+        let expected_methods = DISK_MUTATION_RPC_METHODS.into_iter().map(String::from).collect();
+        assert_eq!(
+            covered_methods, expected_methods,
+            "the disk mutation exclusion set must exactly match handlers exercised by the independent digest test",
+        );
     }
 
     #[tokio::test]
@@ -4774,6 +4925,128 @@ mod tests {
         assert_eq!(signal_response.error_info.as_deref(), Some("unsupported service signal: 99"));
     }
 
+    #[tokio::test]
+    async fn signal_service_body_digest_gate_runs_before_request_handling() {
+        let service = create_test_node_service();
+        let mut vars = HashMap::new();
+        vars.insert(PEER_RESTSIGNAL.to_string(), "99".to_string());
+        vars.insert(PEER_RESTSUB_SYS.to_string(), "scanner".to_string());
+        vars.insert(PEER_RESTDRY_RUN.to_string(), "false".to_string());
+        let message = SignalServiceRequest {
+            vars: Some(Mss { value: vars }),
+        };
+
+        let mut other = message.clone();
+        other
+            .vars
+            .as_mut()
+            .expect("signal vars should exist")
+            .value
+            .insert(PEER_RESTSIGNAL.to_string(), "1".to_string());
+        let mut tampered = Request::new(message.clone());
+        let other_body = other.canonical_body().expect("small signal request should encode");
+        set_tonic_canonical_body_digest(&mut tampered, &other_body).expect("digest metadata should encode");
+        mark_v2_authenticated(&mut tampered);
+        let error = service
+            .signal_service(tampered)
+            .await
+            .expect_err("a tampered signal request must fail before handler logic");
+        assert_eq!(error.code(), tonic::Code::PermissionDenied);
+
+        let mut signed = Request::new(message);
+        let body = signed.get_ref().canonical_body().expect("small signal request should encode");
+        set_tonic_canonical_body_digest(&mut signed, &body).expect("digest metadata should encode");
+        mark_v2_authenticated(&mut signed);
+        let response = service
+            .signal_service(signed)
+            .await
+            .expect("a correctly body-bound signal request must reach handler logic")
+            .into_inner();
+        assert!(!response.success);
+        assert_eq!(response.error_info.as_deref(), Some("unsupported service signal: 99"));
+    }
+
+    #[tokio::test]
+    async fn every_non_disk_mutation_rejects_a_mismatched_body_digest() {
+        let service = create_test_node_service();
+        let mut covered_methods = HashSet::new();
+
+        macro_rules! assert_tampered {
+            ($method:ident, $message:expr) => {{
+                assert!(
+                    covered_methods.insert(normalized_rpc_method(stringify!($method))),
+                    concat!("duplicate non-disk mutation test for ", stringify!($method)),
+                );
+                let mut request = Request::new($message);
+                set_tonic_canonical_body_digest(&mut request, b"unrelated-canonical-body")
+                    .expect("digest metadata should encode");
+                mark_v2_authenticated(&mut request);
+                let error = service
+                    .$method(request)
+                    .await
+                    .expect_err(concat!(stringify!($method), " must reject a mismatched body digest"));
+                assert_eq!(
+                    error.code(),
+                    tonic::Code::PermissionDenied,
+                    concat!(stringify!($method), " must authenticate before any mutation"),
+                );
+            }};
+        }
+
+        assert_tampered!(heal_bucket, HealBucketRequest::default());
+        assert_tampered!(make_bucket, MakeBucketRequest::default());
+        assert_tampered!(delete_bucket, DeleteBucketRequest::default());
+        assert_tampered!(lock, GenerallyLockRequest::default());
+        assert_tampered!(un_lock, GenerallyLockRequest::default());
+        assert_tampered!(force_un_lock, GenerallyLockRequest::default());
+        assert_tampered!(refresh, GenerallyLockRequest::default());
+        assert_tampered!(lock_batch, BatchGenerallyLockRequest::default());
+        assert_tampered!(un_lock_batch, BatchGenerallyLockRequest::default());
+        assert_tampered!(load_bucket_metadata, LoadBucketMetadataRequest::default());
+        assert_tampered!(delete_bucket_metadata, DeleteBucketMetadataRequest::default());
+        assert_tampered!(delete_policy, DeletePolicyRequest::default());
+        assert_tampered!(load_policy, LoadPolicyRequest::default());
+        assert_tampered!(load_policy_mapping, LoadPolicyMappingRequest::default());
+        assert_tampered!(delete_user, DeleteUserRequest::default());
+        assert_tampered!(delete_service_account, DeleteServiceAccountRequest::default());
+        assert_tampered!(load_user, LoadUserRequest::default());
+        assert_tampered!(load_service_account, LoadServiceAccountRequest::default());
+        assert_tampered!(load_group, LoadGroupRequest::default());
+        assert_tampered!(reload_site_replication_config, ReloadSiteReplicationConfigRequest::default());
+        assert_tampered!(signal_service, SignalServiceRequest::default());
+        assert_tampered!(
+            scanner_activity,
+            ScannerActivityRequest {
+                challenge: vec![7; 16].into(),
+                protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,
+                acknowledge_instance_id: String::new(),
+                acknowledge_dirty_usage_generation: 0,
+            }
+        );
+        assert_tampered!(reload_pool_meta, ReloadPoolMetaRequest::default());
+        assert_tampered!(stop_rebalance, StopRebalanceRequest::default());
+        assert_tampered!(load_rebalance_meta, LoadRebalanceMetaRequest::default());
+        assert_tampered!(start_decommission, StartDecommissionRequest::default());
+        assert_tampered!(cancel_decommission, CancelDecommissionRequest::default());
+        assert_tampered!(clear_decommission, ClearDecommissionRequest::default());
+        assert_tampered!(load_transition_tier_config, LoadTransitionTierConfigRequest::default());
+
+        let body_bound_methods: HashSet<_> = node_service_auth_policies()
+            .into_iter()
+            .filter_map(|(method, policy)| (policy == "body-bound").then_some(method))
+            .collect();
+        let disk_methods: HashSet<_> = DISK_MUTATION_RPC_METHODS.into_iter().map(String::from).collect();
+        assert!(
+            disk_methods.is_subset(&body_bound_methods),
+            "every independently tested disk mutation must remain declared body-bound",
+        );
+        let expected_methods: HashSet<_> = body_bound_methods.difference(&disk_methods).cloned().collect();
+        assert_eq!(
+            covered_methods, expected_methods,
+            "proto body-bound non-disk RPCs must exactly match handlers exercised by mismatch tests",
+        );
+    }
+
     #[tokio::test]
     async fn test_scanner_activity_requires_body_bound_auth_before_storage_lookup() {
         let service = create_test_node_service();
@@ -4838,26 +5111,6 @@ mod tests {
             .expect_err("a signed malformed challenge must fail before storage lookup");
         assert_eq!(malformed_current.code(), tonic::Code::InvalidArgument);
 
-        let mut tampered = Request::new(ScannerActivityRequest {
-            challenge: vec![7; 16].into(),
-            protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,
-            acknowledge_instance_id: String::new(),
-            acknowledge_dirty_usage_generation: 0,
-        });
-        let other = ScannerActivityRequest {
-            challenge: vec![8; 16].into(),
-            ..tampered.get_ref().clone()
-        };
-        let other_canonical =
-            rustfs_protos::canonical_scanner_activity_request_body(&other).expect("scanner activity request should encode");
-        set_tonic_canonical_body_digest(&mut tampered, &other_canonical).expect("digest metadata should encode");
-        mark_v2_authenticated(&mut tampered);
-        let tampered = service
-            .scanner_activity(tampered)
-            .await
-            .expect_err("tampered activity challenge must fail before storage lookup");
-        assert_eq!(tampered.code(), tonic::Code::PermissionDenied);
-
         let mut downgraded = Request::new(ScannerActivityRequest {
             challenge: vec![7; 16].into(),
             protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,