test(internode): extend fallback and transition coverage

Add decode-error metrics for internode msgpack/json compatibility paths, extend the mixed fallback e2e assertions for transitioned multipart partNumber reads, and cover manual transition async status polling plus inactive-owner status behavior.

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -808,13 +808,14 @@ async fn assert_part_number_reader_path(
     key: &str,
     expected_part: &[u8],
     full_object_size: usize,
+    object_class: &str,
     expected_path: &str,
 ) -> TestResult {
     let paths = [INLINE_DIRECT, LEGACY_DUPLEX, EMPTY, REMOTE_TRANSITION];
     let size_bucket = size_bucket(full_object_size);
     let mut before = BTreeMap::<&str, u64>::new();
     for path in paths {
-        before.insert(path, collector.reader_path_total(path, MULTIPART, size_bucket).await);
+        before.insert(path, collector.reader_path_total(path, object_class, size_bucket).await);
     }
     let response = client.get_object().bucket(bucket).key(key).part_number(2).send().await?;
     assert_eq!(
@@ -825,17 +826,17 @@ async fn assert_part_number_reader_path(
     let body = response.body.collect().await?.into_bytes();
     assert_eq!(body.as_ref(), expected_part, "partNumber GET body changed for {key}");
     collector
-        .wait_for_reader_path_total(expected_path, MULTIPART, size_bucket, before[expected_path] + 1)
+        .wait_for_reader_path_total(expected_path, object_class, size_bucket, before[expected_path] + 1)
         .await?;
-    collector.wait_for_reader_paths_to_settle(MULTIPART, size_bucket).await?;
+    collector.wait_for_reader_paths_to_settle(object_class, size_bucket).await?;
     for path in paths {
         if path == expected_path {
             continue;
         }
         assert_eq!(
-            collector.reader_path_total(path, MULTIPART, size_bucket).await,
+            collector.reader_path_total(path, object_class, size_bucket).await,
             before[path],
-            "{READER_PATH_COUNTER}{{path={path}, object_class={MULTIPART}, size_bucket={size_bucket}}} must not advance for partNumber GET {key}; expected {expected_path} only"
+            "{READER_PATH_COUNTER}{{path={path}, object_class={object_class}, size_bucket={size_bucket}}} must not advance for partNumber GET {key}; expected {expected_path} only"
         );
     }
     Ok(())
@@ -1290,6 +1291,7 @@ async fn four_node_inline_fallback_controls() -> TestResult {
         multipart_key,
         &second_part,
         multipart_body.len(),
+        MULTIPART,
         LEGACY_DUPLEX,
     )
     .await?;
@@ -1385,7 +1387,7 @@ async fn four_node_multipart_ignores_disk_compression_fallback() -> TestResult {
         ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, MULTIPART),
     )
     .await?;
-    assert_part_number_reader_path(&collector, &client, bucket, key, &second_part, body.len(), LEGACY_DUPLEX).await?;
+    assert_part_number_reader_path(&collector, &client, bucket, key, &second_part, body.len(), MULTIPART, LEGACY_DUPLEX).await?;
 
     Ok(())
 }
@@ -1428,6 +1430,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
         multipart_key,
         &second_part,
         multipart_body.len(),
+        MULTIPART,
         LEGACY_DUPLEX,
     )
     .await?;
@@ -1517,7 +1520,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
     put_lifecycle_with_transition_retry(&hot_client, bucket).await?;
 
     let key = "transition/mixed-multipart.bin";
-    let (body, _, etag) = put_two_part_multipart(&hot_client, bucket, key).await?;
+    let (body, second_part, etag) = put_two_part_multipart(&hot_client, bucket, key).await?;
     wait_for_transition(&hot_client, bucket, key).await?;
     assert!(
         cold_tier_object_count(&cold_client).await? >= 1,
@@ -1529,6 +1532,8 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
         ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), REMOTE_TRANSITION, REMOTE),
     )
     .await?;
+    assert_part_number_reader_path(&collector, &hot_client, bucket, key, &second_part, body.len(), REMOTE, REMOTE_TRANSITION)
+        .await?;
 
     let encrypted_key = "transition/encrypted-sse.bin";
     let encrypted_body = payload(16 * KIB, 0xAB);

crates/e2e_test/src/reliant/tiering.rs

@@ -70,11 +70,13 @@ const MANUAL_DUE_BUCKET: &str = "ilm7-manual-due";
 const MANUAL_DRY_RUN_BUCKET: &str = "ilm7-manual-dry-run";
 const MANUAL_NOT_DUE_BUCKET: &str = "ilm7-manual-not-due";
 const MANUAL_QUEUE_PRESSURE_BUCKET: &str = "ilm7-manual-queue-pressure";
+const MANUAL_ASYNC_STATUS_BUCKET: &str = "ilm7-manual-async-status";
 const MANUAL_QUEUE_PRESSURE_PREFIX: &str = "manual-queue-pressure/";
 const OBJECT_KEY: &str = "tier/鲁A12345/report.bin";
 const MANUAL_DUE_KEY: &str = "manual-due/report.bin";
 const MANUAL_DRY_RUN_KEY: &str = "manual-dry-run/report.bin";
 const MANUAL_NOT_DUE_KEY: &str = "manual-not-due/report.bin";
+const MANUAL_ASYNC_STATUS_KEY: &str = "manual-async-status/report.bin";
 const CONTENT_TYPE: &str = "application/x-ilm7";
 const USER_META_KEY: &str = "ilm7-origin";
 const USER_META_VAL: &str = "hermetic-transition";
@@ -350,6 +352,16 @@ struct ManualTransitionRunReport {
     truncated_by_duration: bool,
 }
 
+#[derive(Debug, Deserialize)]
+struct ManualTransitionJobStatusResponse {
+    job_id: String,
+    status_endpoint: String,
+    status: String,
+    cancel_requested: bool,
+    failure_reason: Option<String>,
+    report: ManualTransitionRunReport,
+}
+
 async fn manual_transition_run(
     hot: &RustFSTestEnvironment,
     bucket: &str,
@@ -379,6 +391,66 @@ async fn manual_transition_run_with_max(
     Ok(serde_json::from_str(&body)?)
 }
 
+async fn manual_transition_async_run(
+    hot: &RustFSTestEnvironment,
+    bucket: &str,
+    prefix: &str,
+    dry_run: bool,
+    max_objects: u64,
+) -> Result<ManualTransitionRunResponse, Box<dyn std::error::Error + Send + Sync>> {
+    let bucket = urlencoding::encode(bucket);
+    let prefix = urlencoding::encode(prefix);
+    let tier = urlencoding::encode(TIER_NAME);
+    let path = format!(
+        "/rustfs/admin/v3/ilm/transition/run?bucket={bucket}&prefix={prefix}&tier={tier}&dryRun={dry_run}&maxObjects={max_objects}&mode=async"
+    );
+    let (status, body) = signed_admin_request(&hot.url, Method::POST, &path, None, &hot.access_key, &hot.secret_key).await?;
+    assert_eq!(status, reqwest::StatusCode::ACCEPTED, "async manual transition response: {body}");
+    Ok(serde_json::from_str(&body)?)
+}
+
+async fn manual_transition_job_status(
+    hot: &RustFSTestEnvironment,
+    status_endpoint: &str,
+) -> Result<ManualTransitionJobStatusResponse, Box<dyn std::error::Error + Send + Sync>> {
+    let (status, body) =
+        signed_admin_request(&hot.url, Method::GET, status_endpoint, None, &hot.access_key, &hot.secret_key).await?;
+    assert_eq!(status, reqwest::StatusCode::OK, "manual transition job status response: {body}");
+    Ok(serde_json::from_str(&body)?)
+}
+
+async fn manual_transition_job_cancel(
+    hot: &RustFSTestEnvironment,
+    status_endpoint: &str,
+) -> Result<ManualTransitionJobStatusResponse, Box<dyn std::error::Error + Send + Sync>> {
+    let (status, body) =
+        signed_admin_request(&hot.url, Method::DELETE, status_endpoint, None, &hot.access_key, &hot.secret_key).await?;
+    assert_eq!(status, reqwest::StatusCode::OK, "manual transition job cancel response: {body}");
+    Ok(serde_json::from_str(&body)?)
+}
+
+async fn wait_for_manual_transition_job_terminal(
+    hot: &RustFSTestEnvironment,
+    status_endpoint: &str,
+    deadline: StdDuration,
+) -> Result<ManualTransitionJobStatusResponse, Box<dyn std::error::Error + Send + Sync>> {
+    let start = Instant::now();
+    loop {
+        let status = manual_transition_job_status(hot, status_endpoint).await?;
+        if matches!(status.status.as_str(), "completed" | "partial" | "cancelled" | "failed" | "unknown") {
+            return Ok(status);
+        }
+        if start.elapsed() >= deadline {
+            return Err(format!(
+                "manual transition job at {status_endpoint} did not reach a terminal state within {}s; last={status:#?}",
+                deadline.as_secs()
+            )
+            .into());
+        }
+        tokio::time::sleep(StdDuration::from_millis(250)).await;
+    }
+}
+
 /// Number of objects currently stored in the cold-tier bucket.
 async fn cold_tier_object_count(cold_client: &Client) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
     let resp = cold_client.list_objects_v2().bucket(TIER_BUCKET).send().await?;
@@ -661,6 +733,86 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_manual_transition_async_job_status_polling() -> TestResult {
+    let mut cold = RustFSTestEnvironment::new().await?;
+    cold.access_key = "manualasynccoldtieradmin".to_string();
+    cold.secret_key = "manualasynccoldtiersecret".to_string();
+    cold.start_rustfs_server_without_cleanup(vec![]).await?;
+    let cold_client = cold.create_s3_client();
+    cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
+
+    let mut hot = RustFSTestEnvironment::new().await?;
+    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
+        .await?;
+    let hot_client = hot.create_s3_client();
+    add_rustfs_tier(&hot, &cold).await?;
+
+    hot_client.create_bucket().bucket(MANUAL_ASYNC_STATUS_BUCKET).send().await?;
+    put_lifecycle_transition_rule(&hot_client, MANUAL_ASYNC_STATUS_BUCKET, "manual-async-status", "manual-async-status/", 1)
+        .await?;
+    put_single_part_object(
+        &hot_client,
+        MANUAL_ASYNC_STATUS_BUCKET,
+        MANUAL_ASYNC_STATUS_KEY,
+        b"manual async status not-yet-due dry-run object",
+    )
+    .await?;
+
+    let before_dry_run_remote_count = cold_tier_object_count(&cold_client).await?;
+    let accepted = manual_transition_async_run(&hot, MANUAL_ASYNC_STATUS_BUCKET, "manual-async-status/", true, 10).await?;
+    assert_eq!(accepted.state, "accepted");
+    assert_eq!(accepted.mode, "durable_job");
+    assert_eq!(accepted.report.bucket, MANUAL_ASYNC_STATUS_BUCKET);
+    assert_eq!(accepted.report.prefix, "manual-async-status/");
+    assert!(accepted.report.dry_run);
+    assert_eq!(accepted.report.scanned, 0);
+    assert_eq!(accepted.report.eligible, 0);
+    let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?;
+    let status_endpoint = accepted
+        .status_endpoint
+        .as_deref()
+        .ok_or("async response must include status_endpoint")?;
+    assert!(
+        status_endpoint.ends_with(job_id),
+        "status endpoint must embed job id: endpoint={status_endpoint}, job_id={job_id}"
+    );
+
+    let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?;
+    assert_eq!(terminal.job_id, job_id);
+    assert_eq!(terminal.status_endpoint, status_endpoint);
+    assert_eq!(terminal.status, "completed", "terminal job response: {terminal:#?}");
+    assert!(!terminal.cancel_requested);
+    assert_eq!(terminal.failure_reason, None);
+    assert_eq!(terminal.report.bucket, MANUAL_ASYNC_STATUS_BUCKET);
+    assert_eq!(terminal.report.prefix, "manual-async-status/");
+    assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME));
+    assert!(terminal.report.dry_run);
+    assert!(terminal.report.lifecycle_config_found);
+    assert_eq!(terminal.report.scanned, 1, "terminal job response: {terminal:#?}");
+    assert_eq!(terminal.report.eligible, 0, "terminal job response: {terminal:#?}");
+    assert_eq!(terminal.report.dry_run_eligible, 0, "terminal job response: {terminal:#?}");
+    assert_eq!(terminal.report.enqueued, 0, "terminal job response: {terminal:#?}");
+    assert_eq!(terminal.report.skipped_not_transition, 1, "terminal job response: {terminal:#?}");
+    assert_eq!(terminal.report.skipped_queue_full, 0);
+    assert_eq!(terminal.report.skipped_queue_closed, 0);
+    assert_eq!(terminal.report.skipped_queue_timeout, 0);
+    assert!(!terminal.report.truncated_by_limit);
+    assert!(!terminal.report.truncated_by_duration);
+
+    let after_cancel = manual_transition_job_cancel(&hot, status_endpoint).await?;
+    assert_eq!(after_cancel.status, "completed");
+    assert!(!after_cancel.cancel_requested);
+    assert_eq!(
+        cold_tier_object_count(&cold_client).await?,
+        before_dry_run_remote_count,
+        "not-yet-due async dry-run job must not write to cold tier"
+    );
+    assert_not_transitioned(&hot_client, MANUAL_ASYNC_STATUS_BUCKET, MANUAL_ASYNC_STATUS_KEY).await?;
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestResult {
     let mut cold = RustFSTestEnvironment::new().await?;

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

@@ -1118,13 +1118,19 @@ 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(Error::from);
+        return T::deserialize(&mut deserializer).map_err(|err| {
+            crate::cluster::rpc::runtime_sources::record_response_msgpack_decode_error(value_name);
+            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(Error::from)
+    serde_json::from_str(json).map_err(|err| {
+        crate::cluster::rpc::runtime_sources::record_response_json_decode_error(value_name);
+        Error::from(err)
+    })
 }
 
 /// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads
@@ -1172,8 +1178,10 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint:
     }
     let mut read_multiple_resps = Vec::with_capacity(response.read_multiple_resps.len());
     for (index, json_str) in response.read_multiple_resps.iter().enumerate() {
-        let resp = serde_json::from_str::<ReadMultipleResp>(json_str)
-            .map_err(|err| Error::other(format!("decode ReadMultipleResp json item {index} from {endpoint} failed: {err}")))?;
+        let resp = serde_json::from_str::<ReadMultipleResp>(json_str).map_err(|err| {
+            crate::cluster::rpc::runtime_sources::record_response_json_decode_error("ReadMultipleResp");
+            Error::other(format!("decode ReadMultipleResp json item {index} from {endpoint} failed: {err}"))
+        })?;
         read_multiple_resps.push(resp);
     }
 
@@ -1221,6 +1229,7 @@ fn decode_batch_read_version_response_items(
     let mut batch_read_version_resps = Vec::with_capacity(response.batch_read_version_resps.len());
     for (index, json_str) in response.batch_read_version_resps.iter().enumerate() {
         let resp = serde_json::from_str::<BatchReadVersionResp>(json_str).map_err(|err| {
+            crate::cluster::rpc::runtime_sources::record_response_json_decode_error("BatchReadVersionResp");
             Error::other(format!("decode BatchReadVersionResp json item {index} from {endpoint} failed: {err}"))
         })?;
         if resp.success {
@@ -3227,12 +3236,15 @@ mod tests {
             ],
             error: None,
         };
+        let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         let err = decode_read_multiple_response_items(response, &endpoint).expect_err("corrupt msgpack item should fail");
+        let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
         let err = err.to_string();
 
         assert!(err.contains("ReadMultipleResp msgpack item 1"), "unexpected error: {err}");
         assert!(err.contains("server:9000"), "unexpected error: {err}");
+        assert!(after > before, "corrupt response msgpack should increment decode-error metrics");
     }
 
     #[test]
@@ -3247,12 +3259,15 @@ mod tests {
             read_multiple_resps_bin: Vec::new(),
             error: None,
         };
+        let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         let err = decode_read_multiple_response_items(response, &endpoint).expect_err("corrupt json item should fail");
+        let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
         let err = err.to_string();
 
         assert!(err.contains("ReadMultipleResp json item 1"), "unexpected error: {err}");
         assert!(err.contains("server:9000"), "unexpected error: {err}");
+        assert!(after > before, "corrupt response JSON should increment decode-error metrics");
     }
 
     fn sample_batch_read_version_resp(index: usize, path: &str, success: bool) -> BatchReadVersionResp {
@@ -3325,13 +3340,16 @@ mod tests {
             ],
             error: None,
         };
+        let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         let err = decode_batch_read_version_response_items(response, &endpoint)
             .expect_err("corrupt msgpack item should fail")
             .to_string();
+        let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         assert!(err.contains("BatchReadVersionResp msgpack item 1"), "unexpected error: {err}");
         assert!(err.contains("server:9000"), "unexpected error: {err}");
+        assert!(after > before, "corrupt batch response msgpack should increment decode-error metrics");
     }
 
     #[test]
@@ -3346,13 +3364,16 @@ mod tests {
             batch_read_version_resps_bin: Vec::new(),
             error: None,
         };
+        let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         let err = decode_batch_read_version_response_items(response, &endpoint)
             .expect_err("corrupt json item should fail")
             .to_string();
+        let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         assert!(err.contains("BatchReadVersionResp json item 1"), "unexpected error: {err}");
         assert!(err.contains("server:9000"), "unexpected error: {err}");
+        assert!(after > before, "corrupt batch response JSON should increment decode-error metrics");
     }
 
     #[test]

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

@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use rustfs_io_metrics::internode_metrics::{
-    INTERNODE_MSGPACK_DIRECTION_RESPONSE, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE,
-    INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM,
-    INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
+    INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_RESPONSE,
+    INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_WRITE_ALL,
+    INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_TRANSPORT_BACKEND_GRPC,
+    INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
 };
 
 #[cfg(test)]
@@ -126,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_error(message: &'static str) {
+    global_internode_metrics().record_msgpack_json_decode_error(
+        INTERNODE_MSGPACK_DIRECTION_RESPONSE,
+        message,
+        INTERNODE_MSGPACK_CODEC_MSGPACK,
+    );
+}
+
+pub(crate) fn record_response_json_decode_error(message: &'static str) {
+    global_internode_metrics().record_msgpack_json_decode_error(
+        INTERNODE_MSGPACK_DIRECTION_RESPONSE,
+        message,
+        INTERNODE_MSGPACK_CODEC_JSON,
+    );
+}
+
 #[cfg(test)]
 pub(crate) fn reset_internode_metrics_for_test() {
     global_internode_metrics().reset_for_test();
@@ -135,3 +152,8 @@ pub(crate) fn reset_internode_metrics_for_test() {
 pub(crate) fn internode_metrics_snapshot_for_test() -> InternodeMetricsSnapshot {
     global_internode_metrics().snapshot()
 }
+
+#[cfg(test)]
+pub(crate) fn internode_msgpack_json_decode_error_total_for_test() -> u64 {
+    global_internode_metrics().msgpack_json_decode_error_total_for_test()
+}

crates/io-metrics/src/internode_metrics.rs

@@ -35,6 +35,8 @@ pub const INTERNODE_TRANSPORT_BACKEND_UNKNOWN: &str = "unknown";
 /// peer's request vs a client decoding a peer's response (grpc-optimization P2).
 pub const INTERNODE_MSGPACK_DIRECTION_REQUEST: &str = "request";
 pub const INTERNODE_MSGPACK_DIRECTION_RESPONSE: &str = "response";
+pub const INTERNODE_MSGPACK_CODEC_MSGPACK: &str = "msgpack";
+pub const INTERNODE_MSGPACK_CODEC_JSON: &str = "json";
 
 const OPERATION_LABEL: &str = "operation";
 const BACKEND_LABEL: &str = "backend";
@@ -44,6 +46,7 @@ const DOMINANT_ERROR_LABEL: &str = "dominant_error";
 const HTTP_VERSION_LABEL: &str = "http_version";
 const DIRECTION_LABEL: &str = "direction";
 const MESSAGE_LABEL: &str = "message";
+const CODEC_LABEL: &str = "codec";
 const INTERNODE_OPERATION_SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_sent_bytes_total";
 const INTERNODE_OPERATION_RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_recv_bytes_total";
 const INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_outgoing_total";
@@ -60,6 +63,7 @@ const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str =
 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_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";
 const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total";
 
@@ -167,6 +171,7 @@ pub struct InternodeMetrics {
     operation_http_versions_total: AtomicU64,
     operation_stall_timeouts_total: AtomicU64,
     operation_write_shutdown_errors_total: AtomicU64,
+    msgpack_json_decode_error_total: AtomicU64,
     signature_v1_fallback_total: AtomicU64,
 }
 
@@ -364,6 +369,22 @@ impl InternodeMetrics {
         counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).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!(
+            INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL,
+            DIRECTION_LABEL => direction,
+            MESSAGE_LABEL => message,
+            CODEC_LABEL => codec
+        )
+        .increment(1);
+    }
+
+    #[doc(hidden)]
+    pub fn msgpack_json_decode_error_total_for_test(&self) -> u64 {
+        self.msgpack_json_decode_error_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
@@ -439,6 +460,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_error_total.store(0, Ordering::Relaxed);
         self.signature_v1_fallback_total.store(0, Ordering::Relaxed);
     }
 }
@@ -742,8 +764,14 @@ mod tests {
             INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL,
             "rustfs_system_network_internode_msgpack_json_fallback_total"
         );
+        assert_eq!(
+            INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL,
+            "rustfs_system_network_internode_msgpack_json_decode_error_total"
+        );
         assert_eq!(INTERNODE_MSGPACK_DIRECTION_REQUEST, "request");
         assert_eq!(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "response");
+        assert_eq!(INTERNODE_MSGPACK_CODEC_MSGPACK, "msgpack");
+        assert_eq!(INTERNODE_MSGPACK_CODEC_JSON, "json");
         assert_eq!(
             INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL,
             "rustfs_system_network_internode_signature_v1_fallback_total"
@@ -758,6 +786,27 @@ mod tests {
         metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
     }
 
+    #[test]
+    fn msgpack_json_decode_error_counter_tracks_codec_direction_and_message() {
+        let metrics = InternodeMetrics::default();
+        assert_eq!(metrics.msgpack_json_decode_error_total_for_test(), 0);
+
+        metrics.record_msgpack_json_decode_error(
+            INTERNODE_MSGPACK_DIRECTION_REQUEST,
+            "FileInfo",
+            INTERNODE_MSGPACK_CODEC_MSGPACK,
+        );
+        metrics.record_msgpack_json_decode_error(
+            INTERNODE_MSGPACK_DIRECTION_RESPONSE,
+            "RawFileInfo",
+            INTERNODE_MSGPACK_CODEC_JSON,
+        );
+
+        assert_eq!(metrics.msgpack_json_decode_error_total_for_test(), 2);
+        metrics.reset_for_test();
+        assert_eq!(metrics.msgpack_json_decode_error_total_for_test(), 0);
+    }
+
     #[test]
     fn signature_v1_fallback_counter_updates_snapshot_and_resets() {
         // Instance-local metrics keep this independent of the process-global registry.

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

@@ -50,6 +50,7 @@ gitignored — attach the paired directories to the PR / issue.
 | `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_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) |
 | lock p99 (lock metrics) | P1 head-of-line-blocking win |
 
@@ -86,10 +87,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` across a release window; it 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_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.
 - **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

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

@@ -20,6 +20,7 @@ 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_error_total{direction, message, codec}
 ```
 
 Incremented whenever a decode falls back to the JSON field because the msgpack payload was
@@ -29,6 +30,7 @@ absent.
 - `direction="response"` — a client decoding a peer's response (`cluster/rpc/remote_disk.rs`),
   including the list-level `ReadMultiple` / `BatchReadVersion` fallbacks.
 - `message` — the value name, e.g. `FileInfo`, `RawFileInfo`, `ReadMultipleResp`.
+- `codec` — the failed codec for decode errors: `msgpack` for corrupt non-empty `_bin`, or `json` for corrupt legacy fallback JSON.
 
 ## Stage 0 — Observe (current stage)
 
@@ -47,6 +49,16 @@ Every series must be `0`. A non-zero value means some peer is still emitting an
 `_bin` (an old node, or a message whose sender does not fill `_bin`) — investigate the
 `{direction, message}` label before proceeding.
 
+Decode errors must also stay at zero across the observation window:
+
+```promql
+sum by (direction, message, codec) (
+  increase(rustfs_system_network_internode_msgpack_json_decode_error_total[30d])
+)
+```
+
+A non-zero `codec="msgpack"` series means a peer sent corrupt or incompatible `_bin` bytes; it must fail closed and block convergence. A non-zero `codec="json"` series means the legacy fallback field was corrupt or semantically incompatible; it also blocks convergence and rollback confidence.
+
 Standing alert (keep enabled through all stages):
 
 ```yaml
@@ -57,6 +69,13 @@ Standing alert (keep enabled through all stages):
   annotations:
     summary: "Internode RPC fell back to JSON decode ({{ $labels.direction }}/{{ $labels.message }})"
     description: "A peer sent an empty msgpack _bin payload. Do NOT advance msgpack-only convergence while this fires."
+- alert: InternodeMsgpackJsonDecodeError
+  expr: sum by (direction, message, codec) (increase(rustfs_system_network_internode_msgpack_json_decode_error_total[15m])) > 0
+  for: 5m
+  labels: { severity: warning }
+  annotations:
+    summary: "Internode RPC msgpack/JSON decode failed ({{ $labels.direction }}/{{ $labels.message }}/{{ $labels.codec }})"
+    description: "A peer sent an undecodable msgpack or JSON compatibility payload. Do NOT advance msgpack-only convergence while this fires."
 ```
 
 ## Field → peer-decoder audit

rustfs/src/admin/handlers/ilm_transition.rs

@@ -537,6 +537,15 @@ fn request_manual_transition_job_cancel(job_id: &str) -> Option<ManualTransition
     Some(runtime.record.clone())
 }
 
+fn mark_manual_transition_job_owner_unknown(mut record: ManualTransitionJobRecord) -> ManualTransitionJobRecord {
+    if !record.status.is_terminal() {
+        record.status = ManualTransitionJobStatus::Unknown;
+        record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc()));
+        record.failure_reason = Some("manual transition job owner is not active on this node".to_string());
+    }
+    record
+}
+
 fn remove_manual_transition_job(job_id: &str) {
     let mut jobs = lock_manual_transition_jobs();
     jobs.remove(job_id);
@@ -570,15 +579,10 @@ async fn load_manual_transition_job_record(store: Arc<ECStore>, job_id: &str) ->
     if let Some(record) = in_memory_manual_transition_job_record(job_id) {
         return Ok(record);
     }
-    let Some(mut record) = read_manual_transition_job_record(store, job_id).await? else {
+    let Some(record) = read_manual_transition_job_record(store, job_id).await? else {
         return Err(s3_error!(NoSuchKey, "manual transition job not found"));
     };
-    if !record.status.is_terminal() {
-        record.status = ManualTransitionJobStatus::Unknown;
-        record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc()));
-        record.failure_reason = Some("manual transition job owner is not active on this node".to_string());
-    }
-    Ok(record)
+    Ok(mark_manual_transition_job_owner_unknown(record))
 }
 
 async fn persist_manual_transition_job_update(store: Arc<ECStore>, record: &ManualTransitionJobRecord) -> bool {
@@ -762,10 +766,8 @@ impl Operation for ManualTransitionJobCancelHandler {
                 return Err(s3_error!(NoSuchKey, "manual transition job not found"));
             };
             if !record.status.is_terminal() {
-                record.status = ManualTransitionJobStatus::Unknown;
                 record.cancel_requested = true;
-                record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc()));
-                record.failure_reason = Some("manual transition job owner is not active on this node".to_string());
+                record = mark_manual_transition_job_owner_unknown(record);
                 save_manual_transition_job_record(store, &record).await?;
             }
             record
@@ -1065,6 +1067,47 @@ mod tests {
         assert!(cancel_token.is_cancelled());
     }
 
+    #[test]
+    fn manual_transition_inactive_owner_marks_nonterminal_job_unknown() {
+        let (bucket, options, _run_mode) =
+            parse_manual_transition_query(Some("bucket=data&prefix=logs/&async=true")).expect("async query should parse");
+        let record = new_manual_transition_job_record(
+            "11111111-1111-4111-8111-111111111111".to_string(),
+            &bucket,
+            &options,
+            OffsetDateTime::UNIX_EPOCH,
+        );
+
+        let unknown = mark_manual_transition_job_owner_unknown(record);
+
+        assert_eq!(unknown.status, ManualTransitionJobStatus::Unknown);
+        assert!(unknown.finished_at.is_some());
+        assert_eq!(
+            unknown.failure_reason.as_deref(),
+            Some("manual transition job owner is not active on this node")
+        );
+    }
+
+    #[test]
+    fn manual_transition_inactive_owner_preserves_terminal_job() {
+        let (bucket, options, _run_mode) =
+            parse_manual_transition_query(Some("bucket=data&prefix=logs/&async=true")).expect("async query should parse");
+        let mut record = new_manual_transition_job_record(
+            "11111111-1111-4111-8111-111111111111".to_string(),
+            &bucket,
+            &options,
+            OffsetDateTime::UNIX_EPOCH,
+        );
+        record.status = ManualTransitionJobStatus::Completed;
+        record.finished_at = Some("1970-01-01T00:00:00Z".to_string());
+
+        let completed = mark_manual_transition_job_owner_unknown(record);
+
+        assert_eq!(completed.status, ManualTransitionJobStatus::Completed);
+        assert_eq!(completed.finished_at.as_deref(), Some("1970-01-01T00:00:00Z"));
+        assert_eq!(completed.failure_reason, None);
+    }
+
     #[test]
     fn manual_transition_handler_requires_set_tier_action() {
         let src = include_str!("ilm_transition.rs");

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

@@ -21,8 +21,9 @@ use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 use bytes::Bytes;
 use rustfs_filemeta::FileInfo;
 use rustfs_io_metrics::internode_metrics::{
-    INTERNODE_MSGPACK_DIRECTION_REQUEST, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL,
-    INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics,
+    INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_REQUEST,
+    INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC,
+    global_internode_metrics,
 };
 use rustfs_protos::proto_gen::node_service::*;
 use serde::de::DeserializeOwned;
@@ -41,14 +42,27 @@ 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| DiskError::other(format!("decode {value_name} msgpack failed: {err}")));
+        return T::deserialize(&mut deserializer).map_err(|err| {
+            global_internode_metrics().record_msgpack_json_decode_error(
+                INTERNODE_MSGPACK_DIRECTION_REQUEST,
+                value_name,
+                INTERNODE_MSGPACK_CODEC_MSGPACK,
+            );
+            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| DiskError::other(format!("decode {value_name} failed: {err}")))
+    serde_json::from_str(json).map_err(|err| {
+        global_internode_metrics().record_msgpack_json_decode_error(
+            INTERNODE_MSGPACK_DIRECTION_REQUEST,
+            value_name,
+            INTERNODE_MSGPACK_CODEC_JSON,
+        );
+        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> {
@@ -1191,6 +1205,7 @@ mod tests {
     };
     use crate::storage::storage_api::ReadMultipleResp;
     use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp;
+    use rustfs_io_metrics::internode_metrics::global_internode_metrics;
     use serde::{Deserialize, Serialize};
 
     #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -1314,18 +1329,24 @@ mod tests {
 
     #[test]
     fn decode_msgpack_or_json_fails_closed_on_corrupt_non_empty_msgpack() {
+        let before = global_internode_metrics().msgpack_json_decode_error_total_for_test();
         let err = decode_msgpack_or_json::<SamplePayload>(b"not-msgpack", r#"{"name":"json","count":1}"#, "SamplePayload")
             .expect_err("corrupt non-empty msgpack must not fall back to JSON");
+        let after = global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         assert!(err.to_string().contains("decode SamplePayload msgpack failed"), "unexpected error: {err}");
+        assert!(after > before, "corrupt msgpack should increment decode-error metrics");
     }
 
     #[test]
     fn decode_msgpack_or_json_reports_corrupt_json_item_when_msgpack_absent() {
+        let before = global_internode_metrics().msgpack_json_decode_error_total_for_test();
         let err = decode_msgpack_or_json::<SamplePayload>(&[], "{not-json", "SamplePayload")
             .expect_err("corrupt json item should fail in fallback branch");
+        let after = global_internode_metrics().msgpack_json_decode_error_total_for_test();
 
         assert!(err.to_string().contains("decode SamplePayload failed"), "unexpected error: {err}");
+        assert!(after > before, "corrupt fallback JSON should increment decode-error metrics");
     }
 
     #[test]