perf(ecstore): expose replay cache RPC sources (#5926)

Track accepted replay cache records by gRPC operation and split Lock/Unlock and ReadVersion methods out of grpc_other so hotpath validation can attribute nonce pressure without changing replay protection semantics.

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

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

@@ -40,8 +40,11 @@ use http::{HeaderMap, HeaderValue, Method, Uri};
 use rustfs_credentials::{DEFAULT_SECRET_KEY, RPC_SECRET_REQUIRED_MESSAGE};
 use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token};
 use rustfs_io_metrics::internode_metrics::{
-    INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE,
+    INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, INTERNODE_OPERATION_GRPC_LOCK,
-    INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics,
+    INTERNODE_OPERATION_GRPC_LOCK_BATCH, INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL,
+    INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_REFRESH,
+    INTERNODE_OPERATION_GRPC_UNLOCK, INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, INTERNODE_OPERATION_GRPC_WRITE_ALL,
+    INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics,
 };
 use rustfs_object_data_cache::{MemoryBasis, resolve_effective_memory};
 use rustfs_utils::get_env_bool;
@@ -340,6 +343,7 @@ struct RpcNonceCacheMetrics<'a> {
     expired: usize,
     entries: usize,
     capacity: usize,
+    record_scope: Option<RpcReplayCacheMetricScope<'a>>,
     overflow_scope: Option<RpcReplayCacheMetricScope<'a>>,
 }
 
@@ -350,6 +354,13 @@ fn publish_nonce_cache_metrics(metrics: Option<RpcNonceCacheMetrics<'_>>) {
     let internode_metrics = global_internode_metrics();
     internode_metrics.record_replay_cache_evictions("expired", metrics.expired);
     internode_metrics.record_replay_cache_state(metrics.entries, metrics.capacity);
+    if let Some(scope) = metrics.record_scope {
+        internode_metrics.record_replay_cache_record_for_operation_and_backend_path(
+            scope.operation,
+            scope.backend,
+            scope.rpc_path,
+        );
+    }
     if let Some(scope) = metrics.overflow_scope {
         internode_metrics.record_replay_cache_overflow_for_operation_and_backend_path(
             scope.operation,
@@ -385,6 +396,7 @@ impl RpcNonceCache {
             expired,
             entries: self.nonces.len(),
             capacity: record.capacity,
+            record_scope: None,
             overflow_scope: None,
         };
         if self.nonces.contains(&record.nonce) {
@@ -409,6 +421,7 @@ impl RpcNonceCache {
             Ok(()),
             Some(RpcNonceCacheMetrics {
                 entries: self.nonces.len(),
+                record_scope: Some(record.metric_scope),
                 ..metrics
             }),
         )
@@ -913,7 +926,15 @@ fn tonic_rpc_metric_operation(path: &str) -> &'static str {
     match parse_tonic_rpc_path(path).ok().map(|(_, rpc_method)| rpc_method) {
         Some("ReadAll") => INTERNODE_OPERATION_GRPC_READ_ALL,
         Some("ReadMultiple") => INTERNODE_OPERATION_GRPC_READ_MULTIPLE,
+        Some("ReadVersion") => INTERNODE_OPERATION_GRPC_READ_VERSION,
+        Some("BatchReadVersion") => INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION,
         Some("WriteAll") => INTERNODE_OPERATION_GRPC_WRITE_ALL,
+        Some("Lock") => INTERNODE_OPERATION_GRPC_LOCK,
+        Some("UnLock") => INTERNODE_OPERATION_GRPC_UNLOCK,
+        Some("LockBatch") => INTERNODE_OPERATION_GRPC_LOCK_BATCH,
+        Some("UnLockBatch") => INTERNODE_OPERATION_GRPC_UNLOCK_BATCH,
+        Some("Refresh") => INTERNODE_OPERATION_GRPC_REFRESH,
+        Some("ForceUnLock") => INTERNODE_OPERATION_GRPC_FORCE_UNLOCK,
         _ => INTERNODE_OPERATION_GRPC_OTHER,
     }
 }
@@ -2457,10 +2478,42 @@ mod tests {
             tonic_rpc_metric_operation("/node_service.NodeService/ReadMultiple"),
             INTERNODE_OPERATION_GRPC_READ_MULTIPLE
         );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/ReadVersion"),
+            INTERNODE_OPERATION_GRPC_READ_VERSION
+        );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/BatchReadVersion"),
+            INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION
+        );
         assert_eq!(
             tonic_rpc_metric_operation("/node_service.NodeService/WriteAll"),
             INTERNODE_OPERATION_GRPC_WRITE_ALL
         );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/Lock"),
+            INTERNODE_OPERATION_GRPC_LOCK
+        );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/UnLock"),
+            INTERNODE_OPERATION_GRPC_UNLOCK
+        );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/LockBatch"),
+            INTERNODE_OPERATION_GRPC_LOCK_BATCH
+        );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/UnLockBatch"),
+            INTERNODE_OPERATION_GRPC_UNLOCK_BATCH
+        );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/Refresh"),
+            INTERNODE_OPERATION_GRPC_REFRESH
+        );
+        assert_eq!(
+            tonic_rpc_metric_operation("/node_service.NodeService/ForceUnLock"),
+            INTERNODE_OPERATION_GRPC_FORCE_UNLOCK
+        );
         assert_eq!(
             tonic_rpc_metric_operation("/node_service.NodeService/SignalService"),
             INTERNODE_OPERATION_GRPC_OTHER
@@ -2554,6 +2607,13 @@ mod tests {
         result
     }
 
+    fn check_test_nonce_record_with_metrics<'a>(
+        cache: &mut RpcNonceCache,
+        record: RpcNonceRecord<'a>,
+    ) -> (std::io::Result<()>, Option<RpcNonceCacheMetrics<'a>>) {
+        cache.check_and_record(record)
+    }
+
     fn test_nonce_record(
         nonce: Uuid,
         signed_at: i64,
@@ -2597,6 +2657,48 @@ mod tests {
         assert!(cache.nonces.contains(&nonce_b));
     }
 
+    #[test]
+    fn nonce_cache_metrics_mark_successful_records_only() {
+        let now = Instant::now();
+        let expiry = now.checked_add(REPLAY_CACHE_RETENTION).expect("test expiry should fit");
+        let nonce_a = Uuid::new_v4();
+        let nonce_b = Uuid::new_v4();
+        let mut cache = RpcNonceCache::default();
+
+        let (recorded, metrics) =
+            check_test_nonce_record_with_metrics(&mut cache, test_nonce_record(nonce_a, 100, now, 100, expiry, 1));
+        recorded.expect("first nonce should be recorded");
+        let metrics = metrics.expect("successful nonce should publish metrics");
+        let record_scope = metrics.record_scope.expect("successful nonce should carry record scope");
+        assert_eq!(record_scope.operation, INTERNODE_OPERATION_GRPC_READ_ALL);
+        assert_eq!(record_scope.backend, INTERNODE_TRANSPORT_BACKEND_GRPC);
+        assert_eq!(record_scope.rpc_path, "/node_service.NodeService/ReadAll");
+        assert!(metrics.overflow_scope.is_none());
+
+        let (replay, metrics) =
+            check_test_nonce_record_with_metrics(&mut cache, test_nonce_record(nonce_a, 100, now, 100, expiry, 1));
+        assert_eq!(
+            replay.expect_err("duplicate nonce must fail closed").to_string(),
+            "RPC request replay detected"
+        );
+        let metrics = metrics.expect("replay rejection should still publish cache state");
+        assert!(metrics.record_scope.is_none());
+        assert!(metrics.overflow_scope.is_none());
+
+        let (overflow, metrics) =
+            check_test_nonce_record_with_metrics(&mut cache, test_nonce_record(nonce_b, 100, now, 100, expiry, 1));
+        assert_eq!(
+            overflow.expect_err("full cache must fail closed").to_string(),
+            "RPC replay cache capacity exceeded"
+        );
+        let metrics = metrics.expect("overflow should publish cache state");
+        assert!(metrics.record_scope.is_none());
+        let overflow_scope = metrics.overflow_scope.expect("overflow should keep diagnostic scope");
+        assert_eq!(overflow_scope.operation, INTERNODE_OPERATION_GRPC_READ_ALL);
+        assert_eq!(overflow_scope.backend, INTERNODE_TRANSPORT_BACKEND_GRPC);
+        assert_eq!(overflow_scope.rpc_path, "/node_service.NodeService/ReadAll");
+    }
+
     // The `rpc_body_digest_fallback_counter` serial group covers every test that drives (or
     // asserts on) the process-global body-digest fallback counter, so exact-delta assertions
     // cannot race with each other.

crates/io-metrics/src/internode_metrics.rs

@@ -28,6 +28,14 @@ pub const INTERNODE_OPERATION_NS_SCANNER: &str = "ns_scanner";
 pub const INTERNODE_OPERATION_GRPC_READ_ALL: &str = "grpc_read_all";
 pub const INTERNODE_OPERATION_GRPC_WRITE_ALL: &str = "grpc_write_all";
 pub const INTERNODE_OPERATION_GRPC_READ_MULTIPLE: &str = "grpc_read_multiple";
+pub const INTERNODE_OPERATION_GRPC_READ_VERSION: &str = "grpc_read_version";
+pub const INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION: &str = "grpc_batch_read_version";
+pub const INTERNODE_OPERATION_GRPC_LOCK: &str = "grpc_lock";
+pub const INTERNODE_OPERATION_GRPC_UNLOCK: &str = "grpc_unlock";
+pub const INTERNODE_OPERATION_GRPC_LOCK_BATCH: &str = "grpc_lock_batch";
+pub const INTERNODE_OPERATION_GRPC_UNLOCK_BATCH: &str = "grpc_unlock_batch";
+pub const INTERNODE_OPERATION_GRPC_REFRESH: &str = "grpc_refresh";
+pub const INTERNODE_OPERATION_GRPC_FORCE_UNLOCK: &str = "grpc_force_unlock";
 pub const INTERNODE_OPERATION_GRPC_OTHER: &str = "grpc_other";
 pub const INTERNODE_TRANSPORT_BACKEND_TCP_HTTP: &str = "tcp-http";
 pub const INTERNODE_TRANSPORT_BACKEND_GRPC: &str = "grpc";
@@ -78,6 +86,7 @@ const INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL: &str = "rustfs_system_network_inter
 const INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL: &str = "rustfs_system_network_internode_replay_cache_overflow_total";
 const INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL: &str =
     "rustfs_system_network_internode_replay_cache_overflow_by_operation_total";
+const INTERNODE_REPLAY_CACHE_RECORDS_TOTAL: &str = "rustfs_system_network_internode_replay_cache_records_total";
 const INTERNODE_REPLAY_CACHE_ENTRIES: &str = "rustfs_system_network_internode_replay_cache_entries";
 const INTERNODE_REPLAY_CACHE_CAPACITY: &str = "rustfs_system_network_internode_replay_cache_capacity";
 const INTERNODE_REPLAY_CACHE_EVICTIONS_TOTAL: &str = "rustfs_system_network_internode_replay_cache_evictions_total";
@@ -157,6 +166,10 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = &
         name: INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL,
         labels: SERVER_OPERATION_BACKEND_RPC_PATH_LABELS,
     },
+    InternodeOperationMetricDescriptor {
+        name: INTERNODE_REPLAY_CACHE_RECORDS_TOTAL,
+        labels: SERVER_OPERATION_BACKEND_RPC_PATH_LABELS,
+    },
     InternodeOperationMetricDescriptor {
         name: INTERNODE_REPLAY_CACHE_ENTRIES,
         labels: SERVER_LABELS,
@@ -619,6 +632,22 @@ impl InternodeMetrics {
         .increment(1);
     }
 
+    pub fn record_replay_cache_record_for_operation_and_backend_path(
+        &self,
+        operation: &'static str,
+        backend: &'static str,
+        rpc_path: &str,
+    ) {
+        counter!(
+            INTERNODE_REPLAY_CACHE_RECORDS_TOTAL,
+            SERVER_LABEL => current_server_label(),
+            OPERATION_LABEL => operation,
+            BACKEND_LABEL => backend,
+            RPC_PATH_LABEL => rpc_path.to_owned()
+        )
+        .increment(1);
+    }
+
     pub fn record_replay_cache_state(&self, entries: usize, capacity: usize) {
         let entries = usize_to_u64_saturating(entries);
         let capacity = usize_to_u64_saturating(capacity);
@@ -964,7 +993,7 @@ mod tests {
 
     #[test]
     fn operation_metric_descriptors_include_backend_and_operation_labels() {
-        assert_eq!(INTERNODE_OPERATION_METRICS.len(), 20);
+        assert_eq!(INTERNODE_OPERATION_METRICS.len(), 21);
         for metric in &INTERNODE_OPERATION_METRICS[..6] {
             assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
         }
@@ -986,14 +1015,18 @@ mod tests {
             INTERNODE_OPERATION_METRICS[13].labels,
             &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL]
         );
-        for metric in &INTERNODE_OPERATION_METRICS[14..16] {
+        assert_eq!(
+            INTERNODE_OPERATION_METRICS[14].labels,
+            &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL]
+        );
+        for metric in &INTERNODE_OPERATION_METRICS[15..17] {
             assert_eq!(metric.labels, &[SERVER_LABEL]);
         }
-        assert_eq!(INTERNODE_OPERATION_METRICS[16].labels, &[SERVER_LABEL, REASON_LABEL]);
+        assert_eq!(INTERNODE_OPERATION_METRICS[17].labels, &[SERVER_LABEL, REASON_LABEL]);
-        assert_eq!(INTERNODE_OPERATION_METRICS[17].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]);
+        assert_eq!(INTERNODE_OPERATION_METRICS[18].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]);
         // Payload histogram + large-payload counter carry operation+backend labels.
-        assert_eq!(INTERNODE_OPERATION_METRICS[18].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
         assert_eq!(INTERNODE_OPERATION_METRICS[19].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
+        assert_eq!(INTERNODE_OPERATION_METRICS[20].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
     }
 
     #[test]
@@ -1004,6 +1037,14 @@ mod tests {
         assert_eq!(INTERNODE_OPERATION_WALK_DIR, "walk_dir");
         assert_eq!(INTERNODE_OPERATION_GRPC_READ_ALL, "grpc_read_all");
         assert_eq!(INTERNODE_OPERATION_GRPC_WRITE_ALL, "grpc_write_all");
+        assert_eq!(INTERNODE_OPERATION_GRPC_READ_VERSION, "grpc_read_version");
+        assert_eq!(INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, "grpc_batch_read_version");
+        assert_eq!(INTERNODE_OPERATION_GRPC_LOCK, "grpc_lock");
+        assert_eq!(INTERNODE_OPERATION_GRPC_UNLOCK, "grpc_unlock");
+        assert_eq!(INTERNODE_OPERATION_GRPC_LOCK_BATCH, "grpc_lock_batch");
+        assert_eq!(INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, "grpc_unlock_batch");
+        assert_eq!(INTERNODE_OPERATION_GRPC_REFRESH, "grpc_refresh");
+        assert_eq!(INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, "grpc_force_unlock");
         assert_eq!(INTERNODE_OPERATION_GRPC_OTHER, "grpc_other");
 
         assert_eq!(INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, "tcp-http");
@@ -1048,26 +1089,30 @@ mod tests {
         );
         assert_eq!(
             INTERNODE_OPERATION_METRICS[14].name,
-            "rustfs_system_network_internode_replay_cache_entries"
+            "rustfs_system_network_internode_replay_cache_records_total"
         );
         assert_eq!(
             INTERNODE_OPERATION_METRICS[15].name,
-            "rustfs_system_network_internode_replay_cache_capacity"
+            "rustfs_system_network_internode_replay_cache_entries"
         );
         assert_eq!(
             INTERNODE_OPERATION_METRICS[16].name,
-            "rustfs_system_network_internode_replay_cache_evictions_total"
+            "rustfs_system_network_internode_replay_cache_capacity"
         );
         assert_eq!(
             INTERNODE_OPERATION_METRICS[17].name,
-            "rustfs_system_storage_erasure_write_quorum_failures_total"
+            "rustfs_system_network_internode_replay_cache_evictions_total"
         );
         assert_eq!(
             INTERNODE_OPERATION_METRICS[18].name,
-            "rustfs_system_network_internode_operation_payload_bytes"
+            "rustfs_system_storage_erasure_write_quorum_failures_total"
         );
         assert_eq!(
             INTERNODE_OPERATION_METRICS[19].name,
+            "rustfs_system_network_internode_operation_payload_bytes"
+        );
+        assert_eq!(
+            INTERNODE_OPERATION_METRICS[20].name,
             "rustfs_system_network_internode_operation_large_payloads_total"
         );
         assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple");
@@ -1091,6 +1136,10 @@ mod tests {
             INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL,
             "rustfs_system_network_internode_signature_v1_fallback_total"
         );
+        assert_eq!(
+            INTERNODE_REPLAY_CACHE_RECORDS_TOTAL,
+            "rustfs_system_network_internode_replay_cache_records_total"
+        );
         assert_eq!(FAILURE_REASON_LABEL, "failure_reason");
         assert_eq!(RPC_PATH_LABEL, "rpc_path");
         assert_eq!(REASON_LABEL, "reason");
@@ -1144,6 +1193,11 @@ mod tests {
                 INTERNODE_TRANSPORT_BACKEND_GRPC,
                 "/node_service.NodeService/ReadAll",
             );
+            metrics.record_replay_cache_record_for_operation_and_backend_path(
+                INTERNODE_OPERATION_GRPC_READ_VERSION,
+                INTERNODE_TRANSPORT_BACKEND_GRPC,
+                "/node_service.NodeService/ReadVersion",
+            );
         });
 
         let snapshot = metrics.snapshot();
@@ -1179,6 +1233,28 @@ mod tests {
         assert_eq!(labels.get(BACKEND_LABEL).map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC));
         assert_eq!(labels.get(RPC_PATH_LABEL).map(String::as_str), Some("/node_service.NodeService/ReadAll"));
         assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty()));
+
+        let records: Vec<_> = entries
+            .iter()
+            .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_REPLAY_CACHE_RECORDS_TOTAL)
+            .collect();
+        assert_eq!(records.len(), 1);
+        let labels: HashMap<_, _> = records[0]
+            .0
+            .key()
+            .labels()
+            .map(|label| (label.key().to_string(), label.value().to_string()))
+            .collect();
+        assert_eq!(
+            labels.get(OPERATION_LABEL).map(String::as_str),
+            Some(INTERNODE_OPERATION_GRPC_READ_VERSION)
+        );
+        assert_eq!(labels.get(BACKEND_LABEL).map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC));
+        assert_eq!(
+            labels.get(RPC_PATH_LABEL).map(String::as_str),
+            Some("/node_service.NodeService/ReadVersion")
+        );
+        assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty()));
     }
 
     #[test]