feat(internode): optimize gRPC transport (#4337)

* feat(internode): P0 gRPC transport tuning, message limits, payload metrics Land the P0 subtask from docs/grpc-optimization: close the client-vs-server transport gaps and add instrumentation to size which unary RPCs need channel isolation in P1. Transport tuning (G3): the client `Endpoint` now disables Nagle and raises the HTTP/2 stream/connection flow-control windows to mirror the server socket, so small lock/health RPCs are not batched and larger metadata responses are not throttled by the 64KiB default window. All env-overridable, 0 opts out. Message-size limits (G1): both `NodeServiceClient` and `NodeServiceServer` set max decode/encode size (default 100MiB) instead of tonic's silent 4MiB cap, so a large multi-version xl.meta or aggregated ReadMultiple no longer fails out_of_range. The server limit is set on `NodeServiceServer` before wrapping in the auth `InterceptedService` (the interceptor type does not expose it). Payload instrumentation (P1 prep): ReadAll/ReadMultiple record a payload-size histogram plus a large-payload counter when a response crosses the configured threshold (default 8MiB), feeding alerting on paths that contend with latency-sensitive control-plane traffic on the shared channel. Threshold-only counter, no per-call hot-path log. Verification: cargo check/test on config, io-metrics, ecstore, rustfs; clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(internode): P1 control/bulk gRPC channel isolation (opt-in) Land the P1 subtask from docs/grpc-optimization: physically separate large bytes-carrying unary RPCs from latency-sensitive control-plane RPCs so a big transfer can no longer head-of-line block a lock/health RPC on the shared HTTP/2 connection (G2/G5). Introduce ChannelClass { Control, Bulk } and get_channel_for_class in protos. Control RPCs keep the per-peer connection keyed by the bare address; Bulk RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion, via a new get_bulk_client) are round-robined across a small per-peer bulk pool. Rather than restructuring the global GLOBAL_CONN_MAP (and every consumer), bulk channels are cached under a composite key (addr\0bulk\0idx). The NUL separator cannot appear in a URL, so bulk keys never collide with the control key. This keeps the blast radius small on a consistency-sensitive path. create_new_channel is refactored into build_channel(dial_addr, cache_key) so several physically distinct channels to one peer cache independently while dialing/TLS still use the real address. Gated by RUSTFS_INTERNODE_CHANNEL_ISOLATION (default OFF) so the default build is byte-for-byte the pre-P1 behavior: bulk resolves to the control channel and the switch is a single-env rollback. RUSTFS_INTERNODE_BULK_CHANNELS (default 2, clamped >=1) sizes the pool. On failure, evict_failed_connection drops the whole bulk pool for the peer (round-robin hides which index was used), avoiding half-dead cached channels. Lock RPCs (remote_locker) already use the default Control path, so lock semantics and retry behavior are unchanged. Verification: cargo check/test on config, protos, ecstore, rustfs; new protos tests for bulk key routing and isolation-off passthrough; clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(internode): P2 msgpack/JSON codec observability + encode buffer presizing Land the safe, wire-compatible slice of P2 from docs/grpc-optimization: the observability prerequisite for retiring the redundant JSON fields, plus a codec micro-optimization. No proto/wire-format change; JSON is still dual-written. Internode RPCs today dual-encode each metadata value as both msgpack (`*_bin`) and a JSON compatibility string, and decoders prefer `_bin` with a JSON fallback. Before the JSON fields can ever be dropped (a cross-version change), that fallback must be proven unused in production. Add rustfs_system_network_internode_msgpack_json_fallback_total{direction, message}: incremented whenever a decode falls back to the JSON field because the msgpack payload was absent. Wired into both directions — the client decoding peer responses (remote_disk.rs, incl. the list-level read_multiple/batch fallbacks) and the server decoding peer requests (node_service/disk.rs). This counter must read zero across a release window before send paths stop writing JSON and the proto text fields are reserved/removed (the deferred P2-1 steps). Also pre-size the msgpack encode buffers (Vec::with_capacity(512)) on both sides, eliminating the repeated growth reallocations for typical FileInfo payloads with zero added copy. Full thread_local buffer pooling is deferred: it needs either an extra copy (unclear net win) or a send-path buffer-return lifecycle, to be justified by a codec microbenchmark first. Verification: cargo check/test on io-metrics, ecstore, rustfs; new fallback counter smoke test; existing codec decode tests green; clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * docs(internode): add msgpack/JSON convergence observation runbook Runbook driving the observation-gated retirement of the redundant JSON compatibility fields on internode gRPC metadata RPCs (grpc-optimization P2-1). Documents the shipped fallback counter (rustfs_system_network_internode_msgpack_json_fallback_total{direction,message}), the PromQL to confirm it reads zero across a release window, a standing alert, and the staged flip/rollback procedure (env-gated msgpack-only send, then proto field removal in N+1). Includes the verified field -> peer-decoder audit: only fields whose peer decodes _bin first may be converged. Notes DeleteVersion.opts (DeleteOptions) is NOT convergence-ready — its server handler is not _bin-first and must gain a decode_msgpack_or_json path first. This gates the send-side change so it cannot empty a JSON field an old peer still needs. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(internode): env-gated msgpack-only send + DeleteVersion _bin support (P2-1) Implements the send-side lever for retiring the redundant JSON compatibility fields on internode gRPC metadata RPCs, plus the missing `_bin` support on the delete path that it depends on (grpc-optimization P2-1). Default-off: the base build is byte-for-byte the prior dual-write behavior. Gated msgpack-only send (RUSTFS_INTERNODE_RPC_MSGPACK_ONLY, default false): - New rustfs_protos::internode_rpc_msgpack_only() reads the flag. - Client (remote_disk.rs) compat_json() and server (node_service/disk.rs) compat_response_json() emit an empty JSON string when the flag is on, so only the msgpack _bin payload is sent. The _bin field is always sent; decoders keep the JSON read fallback. Applied only to fields with a confirmed _bin-first peer decoder (WriteMetadata/UpdateMetadata/RenameData file_info, UpdateMetadata opts, ReadOptions, ReadMultipleReq, BatchReadVersionReq; ReadVersion/ReadXL/RenameData responses and the ReadMultiple/BatchReadVersion response lists). - Only enable after the P2 fallback counter has read zero across a release window (see docs/operations/internode-msgpack-json-convergence-runbook.md). Single-env rollback; no wire-format break. DeleteVersion(s) _bin support (prerequisite): - The DeleteVersion/DeleteVersions protos had NO _bin fields. Add additive (backward-compatible) bytes file_info_bin/opts_bin (DeleteVersion) and repeated bytes versions_bin + bytes opts_bin (DeleteVersions); regenerate the checked-in prost struct. - Client dual-writes them; server decodes them _bin-first with JSON fallback. - These delete fields are kept OUT of the msgpack-only set (always dual-write) until their own fallback counter reads zero across a window with the new decoders fully deployed. DeleteVersion.raw_file_info stays JSON-only (no _bin field yet). Verification: cargo check/test on protos, config, ecstore, rustfs (incl. the six delete request handler tests and a compat_json default-path test); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(internode): P3 cluster peer online/offline health metric Land the safe observability core of P3 (grpc-optimization G6/G8): track each internode peer's reachability and expose the offline count, for parity with MinIO's minio_cluster_servers_offline_total. Pure instrumentation — peer selection and quorum are unchanged. - io-metrics: per-peer PeerHealthState { online, consecutive_failures } registry plus record_peer_reachable/record_peer_unreachable. A peer flips offline after N consecutive failures (dial failures or RPC-triggered evictions) and back online on the next successful dial; the count of offline peers is published to the rustfs_cluster_servers_offline_total gauge. - config: RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD (default 3, clamped >= 1). - protos: build_channel marks the peer reachable on a successful dial and unreachable on a dial failure; evict_failed_connection feeds the failure signal too. Keyed by the real peer address, so control and bulk channels to one peer share health state. Deferred (documented in docs/grpc-optimization P3): startup prewarm (no clean topology-ready hook yet), the offline fast-bypass in peer routing (consistency- sensitive; must not change quorum), and idempotent-read-only retry. This commit is observability only. Verification: cargo check/test on io-metrics, config, protos (new peer-health state-machine and threshold-clamp tests); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(internode): P3 control-channel prewarm + self-healing offline bypass Add the remaining P3 connection-lifecycle levers (grpc-optimization G6/G8), both env-gated and default-off so the base build is unchanged. Prewarm (RUSTFS_INTERNODE_PREWARM, default off): RemoteDisk::new spawns a best-effort background dial of the peer's control channel, deduped per peer address, moving the connect cost off the first RPC. Failures fall through to the existing lazy connect + recovery monitor. Offline bypass (RUSTFS_INTERNODE_OFFLINE_BYPASS, default off): remote_disk get_client/get_bulk_client fast-fail a peer already marked offline instead of paying the connect timeout, so the erasure layer proceeds on quorum sooner. This does NOT change quorum. It is self-healing: cluster_peer_should_bypass lets one request per RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS (default 5s) through to recover the peer even with no background monitor, and the recovery monitor's own probe path calls the client directly so it is never bypassed. io-metrics gains cluster_peer_is_offline / cluster_peer_should_bypass (with a per-peer re-probe timestamp). Scope: data path only — remote_locker (lock RPCs, most consistency-sensitive) is left dual-writing/unbypassed as a follow-up. Verification: cargo check/test on io-metrics, config, ecstore (new self-healing bypass tests; all 105 rpc tests green); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * docs(internode): add A/B benchmark runbook for gRPC optimization stages Reproducible before/after collection procedure for grpc-optimization P0–P3. Since every stage is env-gated, before/after is the same binary with different env — no rebuild. Documents, per stage: the exact env toggles (baseline vs enabled column), which existing bench script to run (run_internode_transport_baseline.sh / run_four_node_cluster_failover_bench.sh), the Prometheus metrics to capture, and the acceptance gates from the design docs (e.g. lock p99 down >= 20% for P1, msgpack fallback counter = 0 before enabling P2, correct rustfs_cluster_servers_offline_total for P3). Live runs require a multi-node cluster + load tool + Prometheus scrape and cannot be produced in a single-process sandbox; artifacts land under target/bench (gitignored) and attach to the PR. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(internode): P3-2 lock-path offline bypass + P3-3 idempotent read retry Extend the offline bypass to the lock path and add opt-in retries for idempotent reads (grpc-optimization P3-2/P3-3). Both env-gated and default-off/zero. Offline bypass (lock path): factor the bypass decision into a shared pub(crate) internode_offline_bypass_reason(addr) and call it from remote_locker::get_client too, so lock RPCs to an offline peer fast-fail (letting dsync reach quorum sooner) instead of paying the connect timeout. Does not change quorum; the self-healing re-probe keeps peers recoverable. Gated by RUSTFS_INTERNODE_OFFLINE_BYPASS (default off). Idempotent read retry (P3-3): add execute_read_with_retry — a bounded, exponential-backoff retry for read-only/reentrant RPCs on transient network errors — and route disk_info through it. RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES defaults to 0 (disabled). Write/lock RPCs are never retried (quorum/idempotency safety, per CLAUDE.md); the wrapper requires an Fn closure so only reads that rebuild their request from borrowed inputs qualify. Deferred: grpc.health.v1 (optional ecosystem-compat only; needs a new tonic-health dep and 3-way hybrid-service wiring — internal needs are met by the existing Ping RPC). Verification: cargo check/test on config, ecstore (105 rpc tests green incl. disk_info now via the retry wrapper); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> * feat(scripts): one-click internode gRPC A/B benchmark driver Wrap the per-stage env matrix from the benchmark runbook into scripts/run_internode_grpc_ab_bench.sh: given --stage <p0|p1|p2|p3> and --phase <before|after>, it emits the stage/phase RUSTFS_INTERNODE_* server env to <out-dir>/server-env.sh and runs the right underlying bench (run_internode_transport_baseline.sh for p0/p1/p2, run_four_node_cluster_failover_bench.sh for p3) into a labeled target/bench/internode-transport/<stage>-<phase>/. Passthrough args after `--` reach the underlying bench; --dry-run previews the env + command. The script is explicit that RUSTFS_INTERNODE_* are server env, so for the load-driven stages the operator must restart rustfs with the emitted env before the run; the docker four-node (p3) path exports them for a forwarding compose. shellcheck-clean. Runbook updated with a "One-click driver" section. Co-Authored-By: heihutu <heihutu@gmail.com> * chore(compose): forward RUSTFS_INTERNODE_* into the four-node cluster The four-node local-build compose only forwarded a fixed whitelist of env, so the internode gRPC knobs (grpc-optimization P0-P3) never reached the containers and the A/B bench driver's "after" phase was a no-op. Forward the full RUSTFS_INTERNODE_* set with defaults matching the binary defaults, so leaving them unset is a no-op and the A/B driver can toggle a stage per phase. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(internode): address Copilot review — retry health action + poison-safe peer health Two review nits on #4337: - P3-3 idempotent read retry (remote_disk.rs): execute_read_with_retry ran every attempt through execute_with_timeout_for_op, which hardcodes FailureHealthAction::MarkFailure. So the first transient error could flip the disk faulty and short-circuit the remaining retries, and each attempt over-counted the failure. Route all but the final attempt through execute_with_timeout_for_op_and_health_action with IgnoreFailure; only the last attempt marks faulty/evicts. No default impact (retries default 0). - Peer-health helpers (io-metrics): record_peer_reachable/record_peer_unreachable, cluster_peer_is_offline and cluster_peer_should_bypass early-returned on a poisoned mutex, permanently stalling the offline gauge and bypass state after a single panic. Recover via PoisonError::into_inner(). Verification: cargo check/test on io-metrics + ecstore (105 rpc tests green); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-31 17:28:12 +00:00 · 2026-07-07 05:18:31 +08:00
parent 8f11222a63
commit 6f613317f6
17 changed files with 1571 additions and 66 deletions
@@ -16,7 +16,9 @@ use crate::cluster::rpc::{TONIC_RPC_PREFIX, gen_signature_headers};
 use crate::disk::error::{DiskError, Error as DiskErrorType};
 use crate::runtime::sources as runtime_sources;
 use http::Method;
-use rustfs_protos::{create_new_channel, proto_gen::node_service::node_service_client::NodeServiceClient};
+use rustfs_protos::{
+    ChannelClass, create_new_channel, get_channel_for_class, proto_gen::node_service::node_service_client::NodeServiceClient,
+};
 use std::{error::Error, io::ErrorKind};
 use tonic::{service::interceptor::InterceptedService, transport::Channel};
 use tracing::debug;
@@ -29,21 +31,38 @@ pub async fn node_service_time_out_client(
    addr: &String,
    interceptor: TonicInterceptor,
 ) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
-    // Try to get cached channel
-    let cached_channel = runtime_sources::cached_node_channel(addr).await;
+    // Default to the latency-sensitive control channel; bulk `bytes` RPCs opt in via the
+    // `_for_class` variant below (grpc-optimization P1).
+    node_service_time_out_client_for_class(addr, interceptor, ChannelClass::Control).await
+}

-    let channel = match cached_channel {
-        Some(channel) => {
-            debug!("Using cached gRPC channel for: {}", addr);
-            channel
-        }
-        None => {
-            // No cached connection, create new one
-            create_new_channel(addr).await?
-        }
+/// Build a `NodeServiceClient` bound to the [`ChannelClass`]-appropriate channel for `addr`.
+///
+/// Bulk `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion) pass
+/// [`ChannelClass::Bulk`] so, when channel isolation is enabled, they are physically isolated
+/// from lock/health RPCs; everything else uses [`ChannelClass::Control`]. When isolation is
+/// disabled the two classes resolve to the same cached channel, i.e. legacy behavior.
+pub async fn node_service_time_out_client_for_class(
+    addr: &String,
+    interceptor: TonicInterceptor,
+    class: ChannelClass,
+) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+    let channel = match class {
+        ChannelClass::Control => match runtime_sources::cached_node_channel(addr).await {
+            Some(channel) => {
+                debug!("Using cached gRPC channel for: {}", addr);
+                channel
+            }
+            // No cached connection, create new one.
+            None => create_new_channel(addr).await?,
+        },
+        ChannelClass::Bulk => get_channel_for_class(addr, ChannelClass::Bulk).await?,
    };

-    Ok(NodeServiceClient::with_interceptor(channel, interceptor))
+    let max_message_size = rustfs_protos::internode_rpc_max_message_size();
+    Ok(NodeServiceClient::with_interceptor(channel, interceptor)
+        .max_decoding_message_size(max_message_size)
+        .max_encoding_message_size(max_message_size))
 }

 pub async fn node_service_time_out_client_no_auth(
@@ -14,6 +14,7 @@

 use crate::cluster::rpc::client::{
    TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client,
+    node_service_time_out_client_for_class, node_service_time_out_client_no_auth,
 };
 use crate::cluster::rpc::internode_data_transport::{
    InternodeDataTransport, ReadStreamRequest, WalkDirStreamRequest, WriteStreamRequest,
@@ -39,6 +40,7 @@ use bytes::Bytes;
 use futures::lock::Mutex;
 use metrics::counter;
 use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
+use rustfs_protos::ChannelClass;
 use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_protos::proto_gen::node_service::{
@@ -77,6 +79,8 @@ enum FailureHealthAction {

 const REMOTE_DISK_OPEN_WRITE_MAX_ATTEMPTS: usize = 2;
 const REMOTE_DISK_OPEN_WRITE_RETRY_BACKOFF: Duration = Duration::from_millis(20);
+/// Base backoff for idempotent read-only RPC retries (grpc-optimization P3-3); doubles per attempt.
+const REMOTE_DISK_READ_RETRY_BASE_BACKOFF: Duration = Duration::from_millis(50);
 const ENV_RUSTFS_METADATA_BATCH_READ: &str = "RUSTFS_METADATA_BATCH_READ";
 const LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC: &str = "RUSTFS_BATCH_METADATA_RPC";
 const BATCH_METADATA_RPC_OFF: &str = "off";
@@ -184,6 +188,78 @@ pub struct RemoteDisk {
    data_transport: Arc<dyn InternodeDataTransport>,
 }

+// ── Connection lifecycle (grpc-optimization P3) ──
+
+/// Whether to prewarm the internode control channel in the background at construction (default off).
+fn internode_prewarm_enabled() -> bool {
+    rustfs_utils::get_env_bool(rustfs_config::ENV_INTERNODE_PREWARM, rustfs_config::DEFAULT_INTERNODE_PREWARM)
+}
+
+/// Whether to fast-fail RPCs to peers already marked offline (default off).
+fn internode_offline_bypass_enabled() -> bool {
+    rustfs_utils::get_env_bool(
+        rustfs_config::ENV_INTERNODE_OFFLINE_BYPASS,
+        rustfs_config::DEFAULT_INTERNODE_OFFLINE_BYPASS,
+    )
+}
+
+/// Re-probe interval for the offline bypass (>= 1s).
+fn internode_offline_reprobe_interval() -> Duration {
+    Duration::from_secs(
+        rustfs_utils::get_env_u64(
+            rustfs_config::ENV_INTERNODE_OFFLINE_REPROBE_SECS,
+            rustfs_config::DEFAULT_INTERNODE_OFFLINE_REPROBE_SECS,
+        )
+        .max(1),
+    )
+}
+
+/// If the offline bypass is enabled and `addr` is marked offline, return a reason string to
+/// fast-fail with instead of paying the connect timeout (grpc-optimization P3-2). Self-healing:
+/// one request per re-probe interval is let through so the peer can recover. Shared by the data
+/// path (`remote_disk`) and the lock path (`remote_locker`).
+pub(crate) fn internode_offline_bypass_reason(addr: &str) -> Option<String> {
+    if internode_offline_bypass_enabled()
+        && rustfs_io_metrics::internode_metrics::cluster_peer_should_bypass(addr, internode_offline_reprobe_interval())
+    {
+        return Some(format!("internode peer {addr} offline; fast-fail bypass (P3)"));
+    }
+    None
+}
+
+/// Number of extra attempts for idempotent read-only control-plane RPCs on transient network
+/// failures (grpc-optimization P3-3). `0` (default) disables retries. Write/lock RPCs never retry.
+fn internode_idempotent_read_retries() -> usize {
+    rustfs_utils::get_env_usize(
+        rustfs_config::ENV_INTERNODE_IDEMPOTENT_READ_RETRIES,
+        rustfs_config::DEFAULT_INTERNODE_IDEMPOTENT_READ_RETRIES,
+    )
+}
+
+/// Peers for which a control-channel prewarm has already been triggered, to dedup the N remote
+/// disks that map to a single peer address.
+static PREWARMED_PEERS: std::sync::LazyLock<std::sync::Mutex<std::collections::HashSet<String>>> =
+    std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new()));
+
+/// Best-effort background prewarm of a peer's control channel (grpc-optimization P3-1). Deduped per
+/// peer; a dial failure just falls through to the existing lazy connect and recovery monitor.
+fn spawn_control_channel_prewarm(addr: String) {
+    {
+        let Ok(mut prewarmed) = PREWARMED_PEERS.lock() else {
+            return;
+        };
+        if !prewarmed.insert(addr.clone()) {
+            return;
+        }
+    }
+    tokio::spawn(async move {
+        match node_service_time_out_client_no_auth(&addr).await {
+            Ok(_) => debug!(addr = %addr, "internode control channel prewarmed"),
+            Err(err) => debug!(addr = %addr, error = %err, "internode control channel prewarm failed (best-effort)"),
+        }
+    });
+}
+
 impl RemoteDisk {
    fn recovery_monitor_span(addr: &str, endpoint: &Endpoint) -> tracing::Span {
        tracing::info_span!(
@@ -231,6 +307,12 @@ impl RemoteDisk {
        };
        record_drive_runtime_state(ep, RuntimeDriveHealthState::Online);

+        // P3-1: move the connect cost off the first RPC by prewarming the control channel in the
+        // background. Deduped per peer, best-effort, opt-in.
+        if internode_prewarm_enabled() {
+            spawn_control_channel_prewarm(disk.addr.clone());
+        }
+
        Ok(disk)
    }

@@ -613,6 +695,49 @@ impl RemoteDisk {
        self.execute_with_timeout_for_op("unknown", operation, timeout_duration).await
    }

+    /// Execute an **idempotent, read-only/reentrant** RPC with a bounded number of retries on
+    /// transient network failures, with exponential backoff (grpc-optimization P3-3). Retries
+    /// default to 0 (disabled). MUST NOT be used for write/lock RPCs — those must never auto-retry
+    /// (quorum/idempotency safety). The `operation` closure is re-invoked per attempt, so it must be
+    /// `Fn` (rebuild the request from borrowed inputs, do not move captured state out).
+    async fn execute_read_with_retry<T, F, Fut>(&self, op: &'static str, operation: F, timeout_duration: Duration) -> Result<T>
+    where
+        F: Fn() -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let max_retries = internode_idempotent_read_retries();
+        let mut attempt = 0usize;
+        loop {
+            // Only the final attempt marks the disk faulty / evicts the channel. Earlier retries
+            // ignore the failure, so a transient error cannot flip the disk into a faulty
+            // short-circuit (which would defeat the retry) or over-count failures.
+            let health_action = if attempt >= max_retries {
+                FailureHealthAction::MarkFailure
+            } else {
+                FailureHealthAction::IgnoreFailure
+            };
+            match self
+                .execute_with_timeout_for_op_and_health_action(op, &operation, timeout_duration, health_action)
+                .await
+            {
+                Err(err) if attempt < max_retries && is_network_like_disk_error(&err) => {
+                    attempt += 1;
+                    let backoff = REMOTE_DISK_READ_RETRY_BASE_BACKOFF
+                        .saturating_mul(1u32 << u32::try_from(attempt - 1).unwrap_or(4).min(4));
+                    debug!(
+                        endpoint = %self.endpoint,
+                        addr = %self.addr,
+                        op,
+                        attempt,
+                        "retrying idempotent read-only RPC after transient network error"
+                    );
+                    tokio::time::sleep(backoff).await;
+                }
+                other => return other,
+            }
+        }
+    }
+
    async fn execute_with_timeout_for_op<T, F, Fut>(
        &self,
        op: &'static str,
@@ -804,12 +929,41 @@ impl RemoteDisk {
        }
    }

+    /// P3-2 offline bypass: when enabled and this peer is marked offline, fast-fail instead of
+    /// paying the connect timeout, so the erasure layer proceeds on quorum sooner. This does not
+    /// change quorum. Self-healing — one request per re-probe interval is let through so the peer
+    /// recovers even without a background monitor. The recovery monitor's own probe path calls the
+    /// client directly and is unaffected.
+    fn offline_bypass_error(&self) -> Option<Error> {
+        internode_offline_bypass_reason(&self.addr).map(Error::other)
+    }
+
    async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
+        if let Some(err) = self.offline_bypass_error() {
+            return Err(err);
+        }
        node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
            .await
            .map_err(|err| Error::other(format!("can not get client, err: {err}")))
    }

+    /// Client for large `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion).
+    /// Routes onto the isolated bulk channel pool so large transfers cannot head-of-line block
+    /// lock/health RPCs (grpc-optimization P1). Falls back to the control channel when isolation
+    /// is disabled.
+    async fn get_bulk_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
+        if let Some(err) = self.offline_bypass_error() {
+            return Err(err);
+        }
+        node_service_time_out_client_for_class(
+            &self.addr,
+            TonicInterceptor::Signature(gen_tonic_signature_interceptor()),
+            ChannelClass::Bulk,
+        )
+        .await
+        .map_err(|err| Error::other(format!("can not get client, err: {err}")))
+    }
+
    async fn disk_ref(&self) -> String {
        (*self.id.lock().await)
            .map(|id| id.to_string())
@@ -817,27 +971,56 @@ impl RemoteDisk {
    }
 }

+/// Initial capacity hint (bytes) for msgpack encode buffers, sized to cover a typical single-
+/// version `FileInfo` without repeated growth reallocations. Larger payloads still grow as needed.
+const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
+
 fn encode_msgpack<T: Serialize>(value: &T) -> Result<Vec<u8>> {
-    let mut serializer = rmp_serde::Serializer::new(Vec::new());
+    let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT));
    value.serialize(&mut serializer)?;
    Ok(serializer.into_inner())
 }

+/// JSON compatibility string for a dual-encoded (`_bin` + text) request field. Returns an empty
+/// string when msgpack-only mode is enabled (grpc-optimization P2-1) so the redundant JSON copy is
+/// not sent; otherwise the legacy JSON encoding. Only use for fields whose peer decodes `_bin`
+/// first — the paired `_bin` (msgpack) field must always be sent alongside.
+fn compat_json<T: Serialize>(value: &T) -> Result<String> {
+    if rustfs_protos::internode_rpc_msgpack_only() {
+        return Ok(String::new());
+    }
+    Ok(serde_json::to_string(value)?)
+}
+
 fn encode_msgpack_named<T: Serialize>(value: &T) -> Result<Vec<u8>> {
-    let mut serializer = rmp_serde::Serializer::new(Vec::new()).with_struct_map();
+    let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
    value.serialize(&mut serializer)?;
    Ok(serializer.into_inner())
 }

-fn decode_msgpack_or_json<T: DeserializeOwned>(binary: &[u8], json: &str) -> Result<T> {
+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);
    }

+    // 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)
 }

+/// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads
+/// and falling back to the JSON compatibility strings. Used to size the RPC for the payload
+/// histogram / large-payload alerting (grpc-optimization P0 instrumentation).
+fn read_multiple_response_payload_len(response: &ReadMultipleResponse) -> usize {
+    if !response.read_multiple_resps_bin.is_empty() {
+        response.read_multiple_resps_bin.iter().map(|buf| buf.len()).sum()
+    } else {
+        response.read_multiple_resps.iter().map(|item| item.len()).sum()
+    }
+}
+
 fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint: &Endpoint) -> Result<Vec<ReadMultipleResp>> {
    if !response.read_multiple_resps_bin.is_empty() {
        if !response.read_multiple_resps.is_empty()
@@ -858,7 +1041,7 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint:

        let mut read_multiple_resps = Vec::with_capacity(response.read_multiple_resps_bin.len());
        for (index, buf) in response.read_multiple_resps_bin.iter().enumerate() {
-            let resp = decode_msgpack_or_json::<ReadMultipleResp>(buf, "").map_err(|err| {
+            let resp = decode_msgpack_or_json::<ReadMultipleResp>(buf, "", "ReadMultipleResp").map_err(|err| {
                Error::other(format!("decode ReadMultipleResp msgpack item {index} from {endpoint} failed: {err}"))
            })?;
            read_multiple_resps.push(resp);
@@ -866,6 +1049,10 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint:
        return Ok(read_multiple_resps);
    }

+    // No msgpack payloads present: the whole list fell back to the JSON compatibility field (P2).
+    if !response.read_multiple_resps.is_empty() {
+        crate::cluster::rpc::runtime_sources::record_response_json_fallback("ReadMultipleResp");
+    }
    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)
@@ -899,7 +1086,7 @@ fn decode_batch_read_version_response_items(

        let mut batch_read_version_resps = Vec::with_capacity(response.batch_read_version_resps_bin.len());
        for (index, buf) in response.batch_read_version_resps_bin.iter().enumerate() {
-            let resp = decode_msgpack_or_json::<BatchReadVersionResp>(buf, "").map_err(|err| {
+            let resp = decode_msgpack_or_json::<BatchReadVersionResp>(buf, "", "BatchReadVersionResp").map_err(|err| {
                Error::other(format!("decode BatchReadVersionResp msgpack item {index} from {endpoint} failed: {err}"))
            })?;
            batch_read_version_resps.push(resp);
@@ -907,6 +1094,10 @@ fn decode_batch_read_version_response_items(
        return Ok(batch_read_version_resps);
    }

+    // No msgpack payloads present: the whole list fell back to the JSON compatibility field (P2).
+    if !response.batch_read_version_resps.is_empty() {
+        crate::cluster::rpc::runtime_sources::record_response_json_fallback("BatchReadVersionResp");
+    }
    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| {
@@ -1207,6 +1398,10 @@ impl DiskAPI for RemoteDisk {

        self.execute_with_timeout(
            || async {
+                // `_bin` support for DeleteVersion is new (grpc-optimization P2); always dual-write
+                // JSON + msgpack until its fallback counter has read zero across a release window.
+                let file_info_bin = encode_msgpack(&fi)?;
+                let opts_bin = encode_msgpack(&opts)?;
                let file_info = serde_json::to_string(&fi)?;
                let opts = serde_json::to_string(&opts)?;

@@ -1221,6 +1416,8 @@ impl DiskAPI for RemoteDisk {
                    file_info,
                    force_del_marker,
                    opts,
+                    file_info_bin: file_info_bin.into(),
+                    opts_bin: opts_bin.into(),
                });

                let response = client.delete_version(request).await?.into_inner();
@@ -1256,6 +1453,18 @@ impl DiskAPI for RemoteDisk {
            return vec![Some(DiskError::FaultyDisk); versions.len()];
        }

+        // `_bin` support for DeleteVersions is new (grpc-optimization P2); always dual-write JSON +
+        // msgpack until its fallback counter has read zero across a release window.
+        let opts_bin = match encode_msgpack(&opts) {
+            Ok(opts_bin) => opts_bin,
+            Err(err) => {
+                let mut errors = Vec::with_capacity(versions.len());
+                for _ in 0..versions.len() {
+                    errors.push(Some(Error::other(err.to_string())));
+                }
+                return errors;
+            }
+        };
        let opts = match serde_json::to_string(&opts) {
            Ok(opts) => opts,
            Err(err) => {
@@ -1267,6 +1476,7 @@ impl DiskAPI for RemoteDisk {
            }
        };
        let mut versions_str = Vec::with_capacity(versions.len());
+        let mut versions_bin = Vec::with_capacity(versions.len());
        for file_info_versions in versions.iter() {
            versions_str.push(match serde_json::to_string(file_info_versions) {
                Ok(versions_str) => versions_str,
@@ -1278,6 +1488,16 @@ impl DiskAPI for RemoteDisk {
                    return errors;
                }
            });
+            versions_bin.push(match encode_msgpack(file_info_versions) {
+                Ok(versions_bin) => Bytes::from(versions_bin),
+                Err(err) => {
+                    let mut errors = Vec::with_capacity(versions.len());
+                    for _ in 0..versions.len() {
+                        errors.push(Some(Error::other(err.to_string())));
+                    }
+                    return errors;
+                }
+            });
        }
        let mut client = match self.get_client().await {
            Ok(client) => client,
@@ -1295,6 +1515,8 @@ impl DiskAPI for RemoteDisk {
            volume: volume.to_string(),
            versions: versions_str,
            opts,
+            versions_bin,
+            opts_bin: opts_bin.into(),
        });

        // TODO: use Error not string
@@ -1396,7 +1618,7 @@ impl DiskAPI for RemoteDisk {
            state = "started",
            "Remote disk RPC started"
        );
-        let file_info = serde_json::to_string(&fi)?;
+        let file_info = compat_json(&fi)?;
        let file_info_bin = encode_msgpack(&fi)?;

        self.execute_with_timeout_for_op(
@@ -1469,8 +1691,8 @@ impl DiskAPI for RemoteDisk {
            state = "started",
            "Remote disk RPC started"
        );
-        let file_info = serde_json::to_string(&fi)?;
-        let opts_str = serde_json::to_string(&opts)?;
+        let file_info = compat_json(&fi)?;
+        let opts_str = compat_json(&opts)?;
        let file_info_bin = encode_msgpack(&fi)?;
        let opts_bin = encode_msgpack(opts)?;

@@ -1526,7 +1748,7 @@ impl DiskAPI for RemoteDisk {
            state = "started",
            "Remote disk RPC started"
        );
-        let opts_str = serde_json::to_string(opts)?;
+        let opts_str = compat_json(opts)?;
        let opts_bin = encode_msgpack(opts)?;

        self.execute_with_timeout(
@@ -1551,7 +1773,7 @@ impl DiskAPI for RemoteDisk {
                    return Err(response.error.unwrap_or_default().into());
                }

-                let file_info = decode_msgpack_or_json::<FileInfo>(&response.file_info_bin, &response.file_info)?;
+                let file_info = decode_msgpack_or_json::<FileInfo>(&response.file_info_bin, &response.file_info, "FileInfo")?;

                Ok(file_info)
            },
@@ -1582,7 +1804,7 @@ impl DiskAPI for RemoteDisk {
            state = "started",
            "Remote disk RPC started"
        );
-        let batch_read_version_req = serde_json::to_string(&req)?;
+        let batch_read_version_req = compat_json(&req)?;
        let batch_read_version_req_bin = encode_msgpack(&req)?;

        let batch_result = self
@@ -1591,7 +1813,7 @@ impl DiskAPI for RemoteDisk {
                move || async move {
                    let disk = self.disk_ref().await;
                    let mut client = self
-                        .get_client()
+                        .get_bulk_client()
                        .await
                        .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                    let request = Request::new(BatchReadVersionRequest {
@@ -1688,7 +1910,8 @@ impl DiskAPI for RemoteDisk {
                    return Err(response.error.unwrap_or_default().into());
                }

-                let raw_file_info = decode_msgpack_or_json::<RawFileInfo>(&response.raw_file_info_bin, &response.raw_file_info)?;
+                let raw_file_info =
+                    decode_msgpack_or_json::<RawFileInfo>(&response.raw_file_info_bin, &response.raw_file_info, "RawFileInfo")?;

                Ok(raw_file_info)
            },
@@ -1723,7 +1946,7 @@ impl DiskAPI for RemoteDisk {
        self.execute_with_timeout_for_op(
            "rename_data",
            || async {
-                let file_info = serde_json::to_string(&fi)?;
+                let file_info = compat_json(&fi)?;
                let file_info_bin = encode_msgpack_named(&fi)?;
                let mut client = self
                    .get_client()
@@ -1745,8 +1968,11 @@ impl DiskAPI for RemoteDisk {
                    return Err(response.error.unwrap_or_default().into());
                }

-                let rename_data_resp =
-                    decode_msgpack_or_json::<RenameDataResp>(&response.rename_data_resp_bin, &response.rename_data_resp)?;
+                let rename_data_resp = decode_msgpack_or_json::<RenameDataResp>(
+                    &response.rename_data_resp_bin,
+                    &response.rename_data_resp,
+                    "RenameDataResp",
+                )?;

                Ok(rename_data_resp)
            },
@@ -2241,11 +2467,11 @@ impl DiskAPI for RemoteDisk {

        self.execute_with_timeout(
            || async {
-                let read_multiple_req = serde_json::to_string(&req)?;
+                let read_multiple_req = compat_json(&req)?;
                let read_multiple_req_bin = encode_msgpack(&req)?;
                let disk = self.disk_ref().await;
                let mut client = self
-                    .get_client()
+                    .get_bulk_client()
                    .await
                    .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                let request = Request::new(ReadMultipleRequest {
@@ -2260,6 +2486,10 @@ impl DiskAPI for RemoteDisk {
                    return Err(response.error.unwrap_or_default().into());
                }

+                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_multiple_recv_bytes(
+                    read_multiple_response_payload_len(&response),
+                );
+
                let read_multiple_resps = decode_read_multiple_response_items(response, &self.endpoint)?;

                Ok(read_multiple_resps)
@@ -2288,7 +2518,7 @@ impl DiskAPI for RemoteDisk {
            || async {
                let data_len = data.len();
                let disk = self.disk_ref().await;
-                let mut client = self.get_client().await.map_err(|err| {
+                let mut client = self.get_bulk_client().await.map_err(|err| {
                    crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_error();
                    Error::other(format!("can not get client, err: {err}"))
                })?;
@@ -2339,7 +2569,7 @@ impl DiskAPI for RemoteDisk {
        self.execute_with_timeout(
            || async {
                let disk = self.disk_ref().await;
-                let mut client = self.get_client().await.map_err(|err| {
+                let mut client = self.get_bulk_client().await.map_err(|err| {
                    crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_error();
                    Error::other(format!("can not get client, err: {err}"))
                })?;
@@ -2373,7 +2603,8 @@ impl DiskAPI for RemoteDisk {

    #[tracing::instrument(skip(self))]
    async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
-        self.execute_with_timeout_for_op(
+        // disk_info is idempotent/read-only, so it is eligible for the P3-3 bounded retry.
+        self.execute_read_with_retry(
            "disk_info",
            || async {
                let opts = serde_json::to_string(&opts)?;
@@ -2650,6 +2881,45 @@ mod tests {
        assert_eq!(decoded[0].data, b"fallback");
    }

+    #[test]
+    fn read_multiple_response_payload_len_prefers_msgpack_and_falls_back_to_json() {
+        let bin_a = encode_msgpack(&sample_read_multiple_resp("a", b"binary")).expect("msgpack should encode");
+        let bin_b = encode_msgpack(&sample_read_multiple_resp("b", b"more")).expect("msgpack should encode");
+        let json = serde_json::to_string(&sample_read_multiple_resp("j", b"fallback")).expect("json should encode");
+
+        // When msgpack bins are present, the length is their aggregate size (JSON strings ignored).
+        let with_bin = ReadMultipleResponse {
+            success: true,
+            read_multiple_resps: vec![json.clone()],
+            read_multiple_resps_bin: vec![bin_a.clone().into(), bin_b.clone().into()],
+            error: None,
+        };
+        assert_eq!(read_multiple_response_payload_len(&with_bin), bin_a.len() + bin_b.len());
+
+        // With no msgpack bins, the JSON compatibility strings are summed instead.
+        let json_only = ReadMultipleResponse {
+            success: true,
+            read_multiple_resps: vec![json.clone()],
+            read_multiple_resps_bin: Vec::new(),
+            error: None,
+        };
+        assert_eq!(read_multiple_response_payload_len(&json_only), json.len());
+
+        // An empty response has zero payload.
+        assert_eq!(read_multiple_response_payload_len(&ReadMultipleResponse::default()), 0);
+    }
+
+    #[test]
+    fn compat_json_dual_writes_by_default() {
+        // msgpack-only defaults off, so compat_json returns the JSON encoding (dual-write). The
+        // empty-string (msgpack-only) path is exercised via the env flag in integration, not here,
+        // to keep this test independent of process-global env state.
+        let resp = sample_read_multiple_resp("file", b"data");
+        let json = compat_json(&resp).expect("compat_json should encode");
+        assert!(!json.is_empty());
+        assert_eq!(json, serde_json::to_string(&resp).expect("json should encode"));
+    }
+
    #[test]
    fn read_multiple_response_decode_reports_corrupt_msgpack_item() {
        let endpoint = sample_remote_endpoint();
@@ -77,6 +77,12 @@ impl RemoteClient {
    }

    pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
+        // P3-2 offline bypass (now covering the lock path too): fast-fail a peer already marked
+        // offline instead of paying the connect timeout, so dsync reaches quorum sooner. Does not
+        // change quorum; the self-healing re-probe keeps the peer recoverable.
+        if let Some(reason) = crate::cluster::rpc::remote_disk::internode_offline_bypass_reason(&self.addr) {
+            return Err(LockError::internal(reason));
+        }
        node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
            .await
            .map_err(|err| LockError::internal(format!("can not get client, err: {err}")))
@@ -13,8 +13,9 @@
 // limitations under the License.

 use rustfs_io_metrics::internode_metrics::{
-    INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM,
-    INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_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_TRANSPORT_BACKEND_GRPC,
+    INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
 };

 #[cfg(test)]
@@ -70,6 +71,43 @@ pub(crate) fn record_remote_disk_grpc_read_all_recv_bytes(bytes: usize) {
        INTERNODE_TRANSPORT_BACKEND_GRPC,
        bytes,
    );
+    record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_ALL, bytes);
+}
+
+pub(crate) fn record_remote_disk_grpc_read_multiple_recv_bytes(bytes: usize) {
+    global_internode_metrics().record_recv_bytes_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_READ_MULTIPLE,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        bytes,
+    );
+    record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, bytes);
+}
+
+/// Payload-size threshold (bytes) above which a unary internode gRPC response is counted as a
+/// "large payload" for alerting. Env-overridable via `RUSTFS_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES`.
+fn internode_rpc_large_payload_warn_bytes() -> usize {
+    rustfs_utils::get_env_usize(
+        rustfs_config::ENV_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES,
+        rustfs_config::DEFAULT_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES,
+    )
+}
+
+/// Record the payload size of a completed unary gRPC RPC into the operation histogram, and
+/// flag it as a large payload when it crosses the configured threshold. This instrumentation
+/// sizes which `bytes`-carrying RPCs contend with latency-sensitive control-plane traffic on
+/// the shared channel, feeding the P1 channel-isolation decision (see docs/grpc-optimization).
+fn record_grpc_payload_size(operation: &'static str, bytes: usize) {
+    let metrics = global_internode_metrics();
+    metrics.record_operation_payload_bytes(operation, INTERNODE_TRANSPORT_BACKEND_GRPC, bytes);
+    if bytes >= internode_rpc_large_payload_warn_bytes() {
+        metrics.record_large_operation_payload(operation, INTERNODE_TRANSPORT_BACKEND_GRPC);
+    }
+}
+
+/// Count a client-side response decode that fell back to the JSON compatibility field because the
+/// msgpack `_bin` payload was absent (grpc-optimization P2). `message` is the value name.
+pub(crate) fn record_response_json_fallback(message: &'static str) {
+    global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, message);
 }

 #[cfg(test)]