feat(internode): track msgpack decode traffic (#5309)

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

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -67,6 +67,7 @@ type MetricValues = Arc<Mutex<BTreeMap<String, MetricPointVersions>>>;
 
 const KIB: usize = 1024;
 const READER_PATH_COUNTER: &str = "rustfs_io_get_object_reader_path_by_size_total";
+const MSGPACK_JSON_DECODE_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_total";
 const MSGPACK_JSON_FALLBACK_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_fallback_total";
 const MSGPACK_JSON_DECODE_ERROR_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total";
 const DIRECTION_LABEL: &str = "direction";
@@ -142,6 +143,7 @@ impl VersionState {
 struct OtlpMetricCollector {
     endpoint: String,
     values: MetricValues,
+    decode_values: MetricValues,
     fallback_values: MetricValues,
     decode_error_values: MetricValues,
     task: JoinHandle<()>,
@@ -152,9 +154,11 @@ impl OtlpMetricCollector {
         let listener = TcpListener::bind("127.0.0.1:0").await?;
         let endpoint = format!("http://{}/v1/metrics", listener.local_addr()?);
         let values = Arc::new(Mutex::new(BTreeMap::new()));
+        let decode_values = Arc::new(Mutex::new(BTreeMap::new()));
         let fallback_values = Arc::new(Mutex::new(BTreeMap::new()));
         let decode_error_values = Arc::new(Mutex::new(BTreeMap::new()));
         let task_values = values.clone();
+        let task_decode_values = decode_values.clone();
         let task_fallback_values = fallback_values.clone();
         let task_decode_error_values = decode_error_values.clone();
         let task = tokio::spawn(async move {
@@ -163,6 +167,7 @@ impl OtlpMetricCollector {
                     break;
                 };
                 let values = task_values.clone();
+                let decode_values = task_decode_values.clone();
                 let fallback_values = task_fallback_values.clone();
                 let decode_error_values = task_decode_error_values.clone();
                 tokio::spawn(async move {
@@ -173,6 +178,7 @@ impl OtlpMetricCollector {
                                 handle_metric_export(
                                     request,
                                     values.clone(),
+                                    decode_values.clone(),
                                     fallback_values.clone(),
                                     decode_error_values.clone(),
                                 )
@@ -185,6 +191,7 @@ impl OtlpMetricCollector {
         Ok(Self {
             endpoint,
             values,
+            decode_values,
             fallback_values,
             decode_error_values,
             task,
@@ -225,6 +232,15 @@ impl OtlpMetricCollector {
             .collect()
     }
 
+    async fn msgpack_json_decode_totals(&self) -> BTreeMap<String, u64> {
+        self.decode_values
+            .lock()
+            .await
+            .iter()
+            .map(|(key, points)| (key.clone(), points.values().map(|(_, value)| value).sum()))
+            .collect()
+    }
+
     async fn msgpack_json_decode_error_totals(&self) -> BTreeMap<String, u64> {
         self.decode_error_values
             .lock()
@@ -302,6 +318,7 @@ impl Drop for OtlpMetricCollector {
 async fn handle_metric_export(
     request: Request<Incoming>,
     values: MetricValues,
+    decode_values: MetricValues,
     fallback_values: MetricValues,
     decode_error_values: MetricValues,
 ) -> Result<Response<Full<Bytes>>, Infallible> {
@@ -334,9 +351,11 @@ async fn handle_metric_export(
     match ExportMetricsServiceRequest::decode(payload.as_slice()) {
         Ok(export) => {
             let mut values = values.lock().await;
+            let mut decode_values = decode_values.lock().await;
             let mut fallback_values = fallback_values.lock().await;
             let mut decode_error_values = decode_error_values.lock().await;
             record_reader_path_metrics(&export, &mut values);
+            record_msgpack_decode_metrics(&export, &mut decode_values);
             record_msgpack_fallback_metrics(&export, &mut fallback_values);
             record_msgpack_decode_error_metrics(&export, &mut decode_error_values);
             Ok(response(StatusCode::OK))
@@ -415,6 +434,19 @@ fn msgpack_fallback_metric_key(direction: &str, message: &str) -> String {
     format!("{direction}\u{1f}{message}")
 }
 
+fn record_msgpack_decode_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, MetricPointVersions>) {
+    record_msgpack_counter_metrics(export, MSGPACK_JSON_DECODE_COUNTER, values, |attributes| {
+        let direction = attribute_string(attributes, DIRECTION_LABEL)?;
+        let message = attribute_string(attributes, MESSAGE_LABEL)?;
+        let codec = attribute_string(attributes, CODEC_LABEL)?;
+        Some(msgpack_decode_metric_key(direction, message, codec))
+    });
+}
+
+fn msgpack_decode_metric_key(direction: &str, message: &str, codec: &str) -> String {
+    format!("{direction}\u{1f}{message}\u{1f}{codec}")
+}
+
 fn record_msgpack_decode_error_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, MetricPointVersions>) {
     record_msgpack_counter_metrics(export, MSGPACK_JSON_DECODE_ERROR_COUNTER, values, |attributes| {
         let direction = attribute_string(attributes, DIRECTION_LABEL)?;
@@ -488,6 +520,33 @@ async fn assert_msgpack_fallback_unchanged(
     Ok(())
 }
 
+async fn assert_msgpack_decode_observed(
+    collector: &OtlpMetricCollector,
+    before: &BTreeMap<String, u64>,
+    series: &[(&str, &str)],
+) -> TestResult {
+    let deadline = Instant::now() + Duration::from_secs(20);
+    loop {
+        let after = collector.msgpack_json_decode_totals().await;
+        let missing = series
+            .iter()
+            .filter_map(|(direction, message)| {
+                let key = msgpack_decode_metric_key(direction, message, MSGPACK_CODEC_MSGPACK);
+                let before_value = before.get(&key).copied().unwrap_or_default();
+                let after_value = after.get(&key).copied().unwrap_or_default();
+                (after_value <= before_value).then_some(format!("{direction}/{message}/{}", MSGPACK_CODEC_MSGPACK))
+            })
+            .collect::<Vec<_>>();
+        if missing.is_empty() {
+            return Ok(());
+        }
+        if Instant::now() >= deadline {
+            return Err(format!("timed out waiting for msgpack decode traffic on {missing:?}; totals={after:?}").into());
+        }
+        sleep(Duration::from_millis(100)).await;
+    }
+}
+
 async fn assert_msgpack_decode_errors_unchanged(
     collector: &OtlpMetricCollector,
     before: &BTreeMap<String, u64>,
@@ -521,6 +580,42 @@ fn attribute_string<'a>(attributes: &'a [KeyValue], wanted_key: &str) -> Option<
     })
 }
 
+#[test]
+fn records_msgpack_decode_metric_with_codec_label() {
+    let export = ExportMetricsServiceRequest {
+        resource_metrics: vec![ResourceMetrics {
+            scope_metrics: vec![ScopeMetrics {
+                metrics: vec![Metric {
+                    name: MSGPACK_JSON_DECODE_COUNTER.to_string(),
+                    data: Some(metric::Data::Sum(Sum {
+                        data_points: vec![NumberDataPoint {
+                            attributes: vec![
+                                metric_attribute(DIRECTION_LABEL, FALLBACK_RESPONSE_DIRECTION),
+                                metric_attribute(MESSAGE_LABEL, "ReadMultipleResp"),
+                                metric_attribute(CODEC_LABEL, MSGPACK_CODEC_MSGPACK),
+                            ],
+                            start_time_unix_nano: 7,
+                            time_unix_nano: 11,
+                            value: Some(number_data_point::Value::AsInt(5)),
+                            ..Default::default()
+                        }],
+                        ..Default::default()
+                    })),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+            ..Default::default()
+        }],
+    };
+    let mut values = BTreeMap::new();
+
+    record_msgpack_decode_metrics(&export, &mut values);
+
+    let key = msgpack_decode_metric_key(FALLBACK_RESPONSE_DIRECTION, "ReadMultipleResp", MSGPACK_CODEC_MSGPACK);
+    assert_eq!(values.get(&key).and_then(|points| points.get(&7)).copied(), Some((11, 5)));
+}
+
 #[test]
 fn records_msgpack_decode_error_metric_with_codec_label() {
     let export = ExportMetricsServiceRequest {
@@ -1660,6 +1755,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
     configure_mixed_msgpack_cluster(&mut cluster, &collector)?;
     cluster.start().await?;
 
+    let decode_before = collector.msgpack_json_decode_totals().await;
     let fallback_before = collector.msgpack_json_fallback_totals().await;
     let decode_errors_before = collector.msgpack_json_decode_error_totals().await;
 
@@ -1685,6 +1781,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
         PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX),
     )
     .await?;
+    assert_msgpack_decode_observed(&collector, &decode_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?;
     assert_msgpack_fallback_unchanged(&collector, &fallback_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?;
     assert_msgpack_decode_errors_unchanged(&collector, &decode_errors_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?;
 
@@ -1862,6 +1959,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
     hot.start().await?;
     let hot_client = hot.create_s3_client(0)?;
 
+    let decode_before = collector.msgpack_json_decode_totals().await;
     let fallback_before = collector.msgpack_json_fallback_totals().await;
     let decode_errors_before = collector.msgpack_json_decode_error_totals().await;
 
@@ -1890,6 +1988,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
         PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), REMOTE, REMOTE_TRANSITION),
     )
     .await?;
+    assert_msgpack_decode_observed(&collector, &decode_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?;
 
     let encrypted_key = "transition/encrypted-sse.bin";
     let encrypted_body = payload(16 * KIB, 0xAB);

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

@@ -1131,19 +1131,31 @@ fn encode_msgpack_named<T: Serialize>(value: &T) -> Result<Vec<u8>> {
 fn decode_msgpack_or_json<T: DeserializeOwned>(binary: &[u8], json: &str, value_name: &'static str) -> Result<T> {
     if !binary.is_empty() {
         let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary));
-        return T::deserialize(&mut deserializer).map_err(|err| {
+        return match T::deserialize(&mut deserializer) {
-            crate::cluster::rpc::runtime_sources::record_response_msgpack_decode_error(value_name);
+            Ok(value) => {
-            Error::from(err)
+                crate::cluster::rpc::runtime_sources::record_response_msgpack_decode(value_name);
-        });
+                Ok(value)
+            }
+            Err(err) => {
+                crate::cluster::rpc::runtime_sources::record_response_msgpack_decode_error(value_name);
+                Err(Error::from(err))
+            }
+        };
     }
 
     // The msgpack payload was absent, so fall back to the JSON compatibility field. This branch
     // must read zero across a release window before the redundant JSON fields can be dropped (P2).
     crate::cluster::rpc::runtime_sources::record_response_json_fallback(value_name);
-    serde_json::from_str(json).map_err(|err| {
+    match serde_json::from_str(json) {
-        crate::cluster::rpc::runtime_sources::record_response_json_decode_error(value_name);
+        Ok(value) => {
-        Error::from(err)
+            crate::cluster::rpc::runtime_sources::record_response_json_decode(value_name);
-    })
+            Ok(value)
+        }
+        Err(err) => {
+            crate::cluster::rpc::runtime_sources::record_response_json_decode_error(value_name);
+            Err(Error::from(err))
+        }
+    }
 }
 
 /// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads
@@ -1195,6 +1207,7 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint:
             crate::cluster::rpc::runtime_sources::record_response_json_decode_error("ReadMultipleResp");
             Error::other(format!("decode ReadMultipleResp json item {index} from {endpoint} failed: {err}"))
         })?;
+        crate::cluster::rpc::runtime_sources::record_response_json_decode("ReadMultipleResp");
         read_multiple_resps.push(resp);
     }
 
@@ -1245,6 +1258,7 @@ fn decode_batch_read_version_response_items(
             crate::cluster::rpc::runtime_sources::record_response_json_decode_error("BatchReadVersionResp");
             Error::other(format!("decode BatchReadVersionResp json item {index} from {endpoint} failed: {err}"))
         })?;
+        crate::cluster::rpc::runtime_sources::record_response_json_decode("BatchReadVersionResp");
         if resp.success {
             validate_decoded_file_info(&resp.file_info)?;
         }
@@ -3120,12 +3134,15 @@ mod tests {
             read_multiple_resps_bin: vec![encode_msgpack(&msgpack_resp).expect("msgpack response should encode").into()],
             error: None,
         };
+        let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test();
 
         let decoded = decode_read_multiple_response_items(response, &endpoint).expect("msgpack response should decode");
+        let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test();
 
         assert_eq!(decoded.len(), 1);
         assert_eq!(decoded[0].file, "msgpack");
         assert_eq!(decoded[0].data, b"binary");
+        assert!(after > before, "successful response msgpack decode should increment traffic metrics");
     }
 
     #[test]
@@ -3138,12 +3155,15 @@ mod tests {
             read_multiple_resps_bin: Vec::new(),
             error: None,
         };
+        let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test();
 
         let decoded = decode_read_multiple_response_items(response, &endpoint).expect("json response should decode");
+        let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test();
 
         assert_eq!(decoded.len(), 1);
         assert_eq!(decoded[0].file, "json");
         assert_eq!(decoded[0].data, b"fallback");
+        assert!(after > before, "successful response JSON decode should increment traffic metrics");
     }
 
     #[test]

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

@@ -127,6 +127,22 @@ pub(crate) fn record_response_json_fallback(message: &'static str) {
     global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, message);
 }
 
+pub(crate) fn record_response_msgpack_decode(message: &'static str) {
+    global_internode_metrics().record_msgpack_json_decode(
+        INTERNODE_MSGPACK_DIRECTION_RESPONSE,
+        message,
+        INTERNODE_MSGPACK_CODEC_MSGPACK,
+    );
+}
+
+pub(crate) fn record_response_json_decode(message: &'static str) {
+    global_internode_metrics().record_msgpack_json_decode(
+        INTERNODE_MSGPACK_DIRECTION_RESPONSE,
+        message,
+        INTERNODE_MSGPACK_CODEC_JSON,
+    );
+}
+
 pub(crate) fn record_response_msgpack_decode_error(message: &'static str) {
     global_internode_metrics().record_msgpack_json_decode_error(
         INTERNODE_MSGPACK_DIRECTION_RESPONSE,

crates/io-metrics/src/internode_metrics.rs

@@ -62,6 +62,7 @@ const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str =
     "rustfs_system_network_internode_operation_write_shutdown_errors_total";
 const INTERNODE_OPERATION_PAYLOAD_BYTES: &str = "rustfs_system_network_internode_operation_payload_bytes";
 const INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL: &str = "rustfs_system_network_internode_operation_large_payloads_total";
+const INTERNODE_MSGPACK_JSON_DECODE_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_decode_total";
 const INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_fallback_total";
 const INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total";
 const INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_signature_v1_fallback_total";
@@ -175,6 +176,7 @@ pub struct InternodeMetrics {
     operation_http_versions_total: AtomicU64,
     operation_stall_timeouts_total: AtomicU64,
     operation_write_shutdown_errors_total: AtomicU64,
+    msgpack_json_decode_total: AtomicU64,
     msgpack_json_decode_error_total: AtomicU64,
     signature_v1_fallback_total: AtomicU64,
     body_digest_fallback_total: AtomicU64,
@@ -375,6 +377,17 @@ impl InternodeMetrics {
         counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).increment(1);
     }
 
+    pub fn record_msgpack_json_decode(&self, direction: &'static str, message: &'static str, codec: &'static str) {
+        self.msgpack_json_decode_total.fetch_add(1, Ordering::Relaxed);
+        counter!(
+            INTERNODE_MSGPACK_JSON_DECODE_TOTAL,
+            DIRECTION_LABEL => direction,
+            MESSAGE_LABEL => message,
+            CODEC_LABEL => codec
+        )
+        .increment(1);
+    }
+
     pub fn record_msgpack_json_decode_error(&self, direction: &'static str, message: &'static str, codec: &'static str) {
         self.msgpack_json_decode_error_total.fetch_add(1, Ordering::Relaxed);
         counter!(
@@ -391,6 +404,11 @@ impl InternodeMetrics {
         self.msgpack_json_decode_error_total.load(Ordering::Relaxed)
     }
 
+    #[doc(hidden)]
+    pub fn msgpack_json_decode_total_for_test(&self) -> u64 {
+        self.msgpack_json_decode_total.load(Ordering::Relaxed)
+    }
+
     /// Count an internode gRPC request that was accepted through the legacy constant-target
     /// signature because it carried no v2 auth headers (rolling-upgrade fallback, see
     /// <https://github.com/rustfs/backlog/issues/1327>). Only accepted requests count: rejected
@@ -488,6 +506,7 @@ impl InternodeMetrics {
         self.operation_http_versions_total.store(0, Ordering::Relaxed);
         self.operation_stall_timeouts_total.store(0, Ordering::Relaxed);
         self.operation_write_shutdown_errors_total.store(0, Ordering::Relaxed);
+        self.msgpack_json_decode_total.store(0, Ordering::Relaxed);
         self.msgpack_json_decode_error_total.store(0, Ordering::Relaxed);
         self.signature_v1_fallback_total.store(0, Ordering::Relaxed);
         self.body_digest_fallback_total.store(0, Ordering::Relaxed);
@@ -790,6 +809,10 @@ mod tests {
             "rustfs_system_network_internode_operation_large_payloads_total"
         );
         assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple");
+        assert_eq!(
+            INTERNODE_MSGPACK_JSON_DECODE_TOTAL,
+            "rustfs_system_network_internode_msgpack_json_decode_total"
+        );
         assert_eq!(
             INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL,
             "rustfs_system_network_internode_msgpack_json_fallback_total"
@@ -816,6 +839,19 @@ mod tests {
         metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
     }
 
+    #[test]
+    fn msgpack_json_decode_counter_tracks_codec_direction_and_message() {
+        let metrics = InternodeMetrics::default();
+        assert_eq!(metrics.msgpack_json_decode_total_for_test(), 0);
+
+        metrics.record_msgpack_json_decode(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo", INTERNODE_MSGPACK_CODEC_MSGPACK);
+        metrics.record_msgpack_json_decode(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo", INTERNODE_MSGPACK_CODEC_JSON);
+
+        assert_eq!(metrics.msgpack_json_decode_total_for_test(), 2);
+        metrics.reset_for_test();
+        assert_eq!(metrics.msgpack_json_decode_total_for_test(), 0);
+    }
+
     #[test]
     fn msgpack_json_decode_error_counter_tracks_codec_direction_and_message() {
         let metrics = InternodeMetrics::default();

docs/operations/internode-grpc-benchmark-runbook.md

@@ -51,6 +51,7 @@ gitignored — attach the paired directories to the PR / issue.
 | `rustfs_system_network_internode_operation_payload_bytes` | payload size distribution (P0/P1 sizing) |
 | `rustfs_system_network_internode_operation_large_payloads_total` | large unary RPCs sharing the channel (P1 target) |
 | `rustfs_system_network_internode_dial_avg_time_nanos`, `..._dial_errors_total` | connect cost (P3 prewarm) |
+| `rustfs_system_network_internode_msgpack_json_decode_total{direction,message,codec}` | must be **>0** for each expected P2 series before zero fallback/error readings are meaningful |
 | `rustfs_system_network_internode_msgpack_json_fallback_total{direction,message}` | must be **0** before enabling both msgpack-only gates (P2) |
 | `rustfs_system_network_internode_msgpack_json_decode_error_total{direction,message,codec}` | must be **0** before enabling both msgpack-only gates (P2) |
 | `rustfs_cluster_servers_offline_total` | offline detection correctness (P3 bypass) |
@@ -89,7 +90,7 @@ column, everything else at defaults. Roll a restart between runs.
   `>4 MiB` multi-version `xl.meta` no longer fails `out_of_range`.
 - **P1** — mixed workload (large `ReadAll` + high-frequency `Refresh`). Acceptance gate from
   the design doc: **lock p99 down ≥ 20%** with `RUSTFS_INTERNODE_CHANNEL_ISOLATION=true`.
-- **P2** — observe `msgpack_json_fallback_total` and `msgpack_json_decode_error_total` across a release window; both must stay **0** before flipping both `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true` and `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=true` (see the msgpack convergence runbook). Codec allocation via a `dhat`/`heaptrack` micro-run.
+- **P2** — observe `msgpack_json_decode_total`, `msgpack_json_fallback_total`, and `msgpack_json_decode_error_total` across a release window; every expected `codec="msgpack"` decode series must be **>0**, while fallback and decode-error series must stay **0** before flipping both `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true` and `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=true` (see the msgpack convergence runbook). Codec allocation via a `dhat`/`heaptrack` micro-run.
 - **P3** — cold-start: first cross-node op latency should drop ~one connect RTT with prewarm.
   Failover: `run_four_node_cluster_failover_bench.sh`, kill a node with
   `RUSTFS_INTERNODE_OFFLINE_BYPASS=true`; expect faster failover and a correct
@@ -107,9 +108,7 @@ the gate verdict (pass/fail + measured delta) in the paired directory's `summary
 | **P3 cold-start** | `run_internode_transport_baseline.sh` (fresh cluster, first cross-node op) | first cross-node op latency **drops ~one connect RTT** with `RUSTFS_INTERNODE_PREWARM=true`. | `..._dial_avg_time_nanos`, first-op `..._operation_duration_ms` |
 | **P3 offline** | dedicated *sustained-offline + survivor cross-node access* experiment (the standard four-node failover bench is **not** sensitive to the bypass — quorum holds, `recovery_seconds=0`) | with `RUSTFS_INTERNODE_OFFLINE_BYPASS=true`, survivor cross-node op latency to the downed peer **fast-fails** instead of hanging the dial timeout; `rustfs_cluster_servers_offline_total` = **1** while down, back to **0** after recovery. | `rustfs_cluster_servers_offline_total`, survivor cross-node `..._operation_duration_ms`, `..._dial_errors_total` |
 
-> **P2 is not a throughput gate.** Its acceptance is operational: `msgpack_json_fallback_total`
+> **P2 is not a throughput gate.** Its acceptance is operational: every expected `msgpack_json_decode_total{codec="msgpack"}` series must have non-zero traffic, while `msgpack_json_fallback_total` and `msgpack_json_decode_error_total` must read **0** across a full release window before both msgpack-only env gates are enabled (see the msgpack convergence runbook). Do not benchmark P2 as before/after throughput.
-> must read **0** across a full release window before both msgpack-only env gates are enabled (see
-> the msgpack convergence runbook). Do not benchmark P2 as before/after throughput.
 
 Artifact layout per stage (attach both halves + the diff):
 

docs/operations/internode-msgpack-json-convergence-runbook.md

@@ -20,11 +20,11 @@ otherwise a rolling upgrade with mixed node versions could read an emptied field
 
 ```
 rustfs_system_network_internode_msgpack_json_fallback_total{direction, message}
+rustfs_system_network_internode_msgpack_json_decode_total{direction, message, codec}
 rustfs_system_network_internode_msgpack_json_decode_error_total{direction, message, codec}
 ```
 
-Incremented whenever a decode falls back to the JSON field because the msgpack payload was
+The decode counter increments after a successful msgpack or JSON compatibility decode. The fallback counter increments whenever a decode falls back to the JSON field because the msgpack payload was absent. The decode-error counter increments whenever either codec fails to decode.
-absent.
 
 - `direction="request"` — a server decoding a peer's request (`node_service/disk.rs`).
 - `direction="response"` — a client decoding a peer's response (`cluster/rpc/remote_disk.rs`),
@@ -34,8 +34,18 @@ absent.
 
 ## Stage 0 — Observe (current stage)
 
-Ship the current release (which contains the counter) and let it run for **at least one
+Ship the current release (which contains these counters) and let it run for **at least one
-full release window** across the whole fleet. The counter must stay at **zero**.
+full release window** across the whole fleet. The fallback and decode-error counters must stay at **zero**.
+
+First confirm each expected message/direction has real traffic in the observation window:
+
+```promql
+sum by (direction, message, codec) (
+  increase(rustfs_system_network_internode_msgpack_json_decode_total[30d])
+)
+```
+
+For every convergence-ready message/direction below, the `codec="msgpack"` series must be non-zero before a zero fallback result is meaningful. Missing series, zero traffic, counter reset, or scrape gaps make the gate inconclusive rather than passed.
 
 Confirm zero across the observation window (adjust `[30d]` to the window length):
 

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

@@ -43,27 +43,47 @@ fn decode_msgpack_or_json<T: DeserializeOwned>(
 ) -> std::result::Result<T, DiskError> {
     if !binary.is_empty() {
         let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary));
-        return T::deserialize(&mut deserializer).map_err(|err| {
+        return match T::deserialize(&mut deserializer) {
-            global_internode_metrics().record_msgpack_json_decode_error(
+            Ok(value) => {
-                INTERNODE_MSGPACK_DIRECTION_REQUEST,
+                global_internode_metrics().record_msgpack_json_decode(
-                value_name,
+                    INTERNODE_MSGPACK_DIRECTION_REQUEST,
-                INTERNODE_MSGPACK_CODEC_MSGPACK,
+                    value_name,
-            );
+                    INTERNODE_MSGPACK_CODEC_MSGPACK,
-            DiskError::other(format!("decode {value_name} msgpack failed: {err}"))
+                );
-        });
+                Ok(value)
+            }
+            Err(err) => {
+                global_internode_metrics().record_msgpack_json_decode_error(
+                    INTERNODE_MSGPACK_DIRECTION_REQUEST,
+                    value_name,
+                    INTERNODE_MSGPACK_CODEC_MSGPACK,
+                );
+                Err(DiskError::other(format!("decode {value_name} msgpack failed: {err}")))
+            }
+        };
     }
 
     // The msgpack payload was absent, so fall back to the JSON compatibility field. This branch
     // must read zero across a release window before the redundant JSON fields can be dropped (P2).
     global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, value_name);
-    serde_json::from_str(json).map_err(|err| {
+    match serde_json::from_str(json) {
-        global_internode_metrics().record_msgpack_json_decode_error(
+        Ok(value) => {
-            INTERNODE_MSGPACK_DIRECTION_REQUEST,
+            global_internode_metrics().record_msgpack_json_decode(
-            value_name,
+                INTERNODE_MSGPACK_DIRECTION_REQUEST,
-            INTERNODE_MSGPACK_CODEC_JSON,
+                value_name,
-        );
+                INTERNODE_MSGPACK_CODEC_JSON,
-        DiskError::other(format!("decode {value_name} failed: {err}"))
+            );
-    })
+            Ok(value)
+        }
+        Err(err) => {
+            global_internode_metrics().record_msgpack_json_decode_error(
+                INTERNODE_MSGPACK_DIRECTION_REQUEST,
+                value_name,
+                INTERNODE_MSGPACK_CODEC_JSON,
+            );
+            Err(DiskError::other(format!("decode {value_name} failed: {err}")))
+        }
+    }
 }
 
 fn encode_msgpack<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
@@ -1298,15 +1318,20 @@ mod tests {
         };
 
         let binary = encode_msgpack(&payload, "SamplePayload").unwrap();
+        let before = global_internode_metrics().msgpack_json_decode_total_for_test();
         let decoded =
             decode_msgpack_or_json::<SamplePayload>(&binary, r#"{"name":"ignored","count":1}"#, "SamplePayload").unwrap();
+        let after = global_internode_metrics().msgpack_json_decode_total_for_test();
 
         assert_eq!(decoded, payload);
+        assert!(after > before, "successful request msgpack decode should increment traffic metrics");
     }
 
     #[test]
     fn decode_msgpack_or_json_falls_back_to_json() {
+        let before = global_internode_metrics().msgpack_json_decode_total_for_test();
         let decoded = decode_msgpack_or_json::<SamplePayload>(&[], r#"{"name":"compat","count":7}"#, "SamplePayload").unwrap();
+        let after = global_internode_metrics().msgpack_json_decode_total_for_test();
 
         assert_eq!(
             decoded,
@@ -1315,6 +1340,7 @@ mod tests {
                 count: 7,
             }
         );
+        assert!(after > before, "successful request JSON fallback decode should increment traffic metrics");
     }
 
     fn with_internode_msgpack_env<R>(vars: [(&'static str, Option<&'static str>); 2], f: impl FnOnce() -> R) -> R {