feat(ecstore): add internode transport boundary and TCP baseline runner (#3010)

* docs: add internode data transport RFC

* feat: add internode operation metrics

* fix feedback

* fix(ci): fallback protoc token to github.token

* feat(ecstore): add internode transport boundary and baseline runner

* feat(internode): harden data transport baseline

* Revert "feat(internode): harden data transport baseline"

This reverts commit 5b8d6b8aa4.

* fix(internode): address baseline review comments

* fix(ci): pin setup-protoc to stable release

* fix(ci): install protoc via apt on linux

* fix(ci): restore protoc install for macos and windows

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

.github/actions/setup/action.yml

@@ -53,13 +53,14 @@ runs:
           musl-tools \
           build-essential \
           pkg-config \
-          libssl-dev
+          libssl-dev \
+          protobuf-compiler
 
     - name: Install protoc
       uses: rustfs/setup-protoc@v3.0.1
       with:
-        version: "34.1"
-        repo-token: ${{ inputs.github-token || github.token }}
+        version: "29.3"
+        repo-token: ${{ github.token }}
 
     - name: Install flatc
       uses: Nugine/setup-flatc@v1

INTERNODE_DATA_TRANSPORT_RFC.md

@@ -169,6 +169,7 @@ Existing benchmark assets:
 - `scripts/run_object_batch_bench_enhanced.sh`
 - `scripts/run_object_batch_bench_abc.sh`
 - `scripts/run_four_node_cluster_failover_bench.sh`
+- `scripts/run_internode_transport_baseline.sh` (scenario matrix wrapper for local vs distributed TCP baseline artifacts)
 - Criterion benches under `crates/ecstore/benches/`
 
 These mostly cover S3/object workload or erasure coding performance. They do
@@ -222,6 +223,15 @@ The baseline should produce a machine-readable artifact, for example
 `target/bench/internode-transport/<timestamp>/summary.csv`, plus the exact
 commands and configuration used.
 
+### Baseline runner entry point
+
+Use `scripts/run_internode_transport_baseline.sh` to execute a reproducible
+S3 PUT/GET matrix against `local` and `distributed` scenarios and export:
+
+- `summary.csv` (throughput/latency summary per workload and object size)
+- `internode_metric_deltas.csv` (operation-level internode metric deltas when
+  `--metrics-url` is provided)
+
 ## Transport Abstraction Proposal
 
 ### Design principle

crates/config/src/constants/internode.rs

@@ -32,6 +32,10 @@ pub const DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS: u64 = 3;
 pub const ENV_INTERNODE_RPC_TIMEOUT_SECS: &str = "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS";
 pub const DEFAULT_INTERNODE_RPC_TIMEOUT_SECS: u64 = 10;
 
+/// Environment variable for selecting the internode data-plane transport backend.
+pub const ENV_RUSTFS_INTERNODE_DATA_TRANSPORT: &str = "RUSTFS_INTERNODE_DATA_TRANSPORT";
+pub const DEFAULT_INTERNODE_DATA_TRANSPORT: &str = "tcp-http";
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -58,5 +62,7 @@ mod tests {
             "RUSTFS_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS"
         );
         assert_eq!(ENV_INTERNODE_RPC_TIMEOUT_SECS, "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS");
+        assert_eq!(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, "RUSTFS_INTERNODE_DATA_TRANSPORT");
+        assert_eq!(DEFAULT_INTERNODE_DATA_TRANSPORT, "tcp-http");
     }
 }

crates/ecstore/src/rpc/internode_data_transport.rs

@@ -0,0 +1,165 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::disk::error::Result;
+use crate::disk::{FileReader, FileWriter};
+use crate::rpc::build_auth_headers;
+use async_trait::async_trait;
+use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE};
+use rustfs_config::{DEFAULT_INTERNODE_DATA_TRANSPORT, ENV_RUSTFS_INTERNODE_DATA_TRANSPORT};
+use rustfs_rio::{HttpReader, HttpWriter};
+use std::sync::{Arc, OnceLock};
+use std::time::Duration;
+use tracing::warn;
+
+static INTERNODE_DATA_TRANSPORT: OnceLock<Arc<dyn InternodeDataTransport>> = OnceLock::new();
+
+#[derive(Debug, Clone)]
+pub struct ReadStreamRequest {
+    pub endpoint: String,
+    pub disk: String,
+    pub volume: String,
+    pub path: String,
+    pub offset: usize,
+    pub length: usize,
+}
+
+#[derive(Debug, Clone)]
+pub struct WriteStreamRequest {
+    pub endpoint: String,
+    pub disk: String,
+    pub volume: String,
+    pub path: String,
+    pub append: bool,
+    pub size: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct WalkDirStreamRequest {
+    pub endpoint: String,
+    pub disk: String,
+    pub body: Vec<u8>,
+    pub stall_timeout: Option<Duration>,
+}
+
+#[async_trait]
+pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
+    async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader>;
+    async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter>;
+    async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader>;
+    fn name(&self) -> &'static str;
+}
+
+#[derive(Debug, Default)]
+pub struct TcpHttpInternodeDataTransport;
+
+#[async_trait]
+impl InternodeDataTransport for TcpHttpInternodeDataTransport {
+    async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader> {
+        let url = format!(
+            "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
+            request.endpoint,
+            urlencoding::encode(&request.disk),
+            urlencoding::encode(&request.volume),
+            urlencoding::encode(&request.path),
+            request.offset,
+            request.length
+        );
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        build_auth_headers(&url, &Method::GET, &mut headers)?;
+        Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?))
+    }
+
+    async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter> {
+        let url = format!(
+            "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
+            request.endpoint,
+            urlencoding::encode(&request.disk),
+            urlencoding::encode(&request.volume),
+            urlencoding::encode(&request.path),
+            request.append,
+            request.size
+        );
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        build_auth_headers(&url, &Method::PUT, &mut headers)?;
+        Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?))
+    }
+
+    async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader> {
+        let url = format!("{}/rustfs/rpc/walk_dir?disk={}", request.endpoint, urlencoding::encode(&request.disk));
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        build_auth_headers(&url, &Method::GET, &mut headers)?;
+        Ok(Box::new(
+            HttpReader::new_with_stall_timeout(url, Method::GET, headers, Some(request.body), request.stall_timeout).await?,
+        ))
+    }
+
+    fn name(&self) -> &'static str {
+        DEFAULT_INTERNODE_DATA_TRANSPORT
+    }
+}
+
+fn build_internode_data_transport(configured_transport: Option<&str>) -> Arc<dyn InternodeDataTransport> {
+    match configured_transport.map(str::trim).filter(|transport| !transport.is_empty()) {
+        Some(transport) if transport.eq_ignore_ascii_case(DEFAULT_INTERNODE_DATA_TRANSPORT) => {
+            Arc::new(TcpHttpInternodeDataTransport)
+        }
+        Some(transport) => {
+            warn!(
+                env = ENV_RUSTFS_INTERNODE_DATA_TRANSPORT,
+                requested = %transport,
+                fallback = DEFAULT_INTERNODE_DATA_TRANSPORT,
+                "unknown internode data transport, using default backend"
+            );
+            Arc::new(TcpHttpInternodeDataTransport)
+        }
+        None => Arc::new(TcpHttpInternodeDataTransport),
+    }
+}
+
+pub fn build_internode_data_transport_from_env() -> Arc<dyn InternodeDataTransport> {
+    Arc::clone(
+        INTERNODE_DATA_TRANSPORT
+            .get_or_init(|| build_internode_data_transport(std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok().as_deref())),
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn transport_config_defaults_to_tcp_http() {
+        let transport = build_internode_data_transport(None);
+
+        assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+    }
+
+    #[test]
+    fn transport_config_accepts_tcp_http() {
+        let transport = build_internode_data_transport(Some("TCP-HTTP"));
+
+        assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+    }
+
+    #[test]
+    fn transport_config_falls_back_to_tcp_http_for_unknown_backend() {
+        let transport = build_internode_data_transport(Some("rdma"));
+
+        assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+    }
+}

crates/ecstore/src/rpc/mod.rs

@@ -15,6 +15,7 @@
 mod client;
 mod context_propagation;
 mod http_auth;
+mod internode_data_transport;
 mod peer_rest_client;
 mod peer_s3_client;
 mod remote_disk;

crates/ecstore/src/rpc/remote_disk.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::disk::error::{Error, Result};
 use crate::disk::{
     CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader,
     FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
@@ -27,14 +28,12 @@ use crate::disk::{disk_store::DiskHealthTracker, error::DiskError, local::ScanGu
 use crate::rpc::client::{
     TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client,
 };
-use crate::set_disk::DEFAULT_READ_BUFFER_SIZE;
-use crate::{
-    disk::error::{Error, Result},
-    rpc::build_auth_headers,
+use crate::rpc::internode_data_transport::{
+    InternodeDataTransport, ReadStreamRequest, WalkDirStreamRequest, WriteStreamRequest, build_internode_data_transport_from_env,
 };
+use crate::set_disk::DEFAULT_READ_BUFFER_SIZE;
 use bytes::Bytes;
 use futures::lock::Mutex;
-use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE};
 use metrics::counter;
 use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
 use rustfs_io_metrics::internode_metrics::{
@@ -49,7 +48,6 @@ use rustfs_protos::proto_gen::node_service::{
     RenameFileRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest,
     node_service_client::NodeServiceClient,
 };
-use rustfs_rio::{HttpReader, HttpWriter};
 use serde::{Serialize, de::DeserializeOwned};
 use std::{
     io::Cursor,
@@ -109,6 +107,7 @@ pub struct RemoteDisk {
     health: Arc<DiskHealthTracker>,
     /// Cancellation token for monitoring tasks
     cancel_token: CancellationToken,
+    data_transport: Arc<dyn InternodeDataTransport>,
 }
 
 impl RemoteDisk {
@@ -130,6 +129,7 @@ impl RemoteDisk {
             health_check: opt.health_check && env_health_check,
             health: Arc::new(DiskHealthTracker::new()),
             cancel_token: CancellationToken::new(),
+            data_transport: build_internode_data_transport_from_env(),
         };
         record_drive_runtime_state(ep, RuntimeDriveHealthState::Online);
 
@@ -1200,23 +1200,16 @@ impl DiskAPI for RemoteDisk {
             "walk_dir",
             || async {
                 let disk = self.disk_ref().await;
-
-                let url = format!("{}/rustfs/rpc/walk_dir?disk={}", self.endpoint.grid_host(), urlencoding::encode(&disk),);
-
                 let opts = serde_json::to_vec(&opts)?;
-
-                let mut headers = HeaderMap::new();
-                headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-                build_auth_headers(&url, &Method::GET, &mut headers)?;
-
-                let mut reader = HttpReader::new_with_stall_timeout(
-                    url,
-                    Method::GET,
-                    headers,
-                    Some(opts),
-                    Some(get_drive_walkdir_stall_timeout()),
-                )
-                .await?;
+                let mut reader = self
+                    .data_transport
+                    .open_walk_dir(WalkDirStreamRequest {
+                        endpoint: self.endpoint.grid_host(),
+                        disk,
+                        body: opts,
+                        stall_timeout: Some(get_drive_walkdir_stall_timeout()),
+                    })
+                    .await?;
 
                 copy_stream_with_buffer(&mut reader, wr, DEFAULT_READ_BUFFER_SIZE).await?;
 
@@ -1248,21 +1241,16 @@ impl DiskAPI for RemoteDisk {
             return Err(DiskError::FaultyDisk);
         }
         let disk = self.disk_ref().await;
-
-        let url = format!(
-            "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
-            self.endpoint.grid_host(),
-            urlencoding::encode(&disk),
-            urlencoding::encode(volume),
-            urlencoding::encode(path),
-            offset,
-            length
-        );
-
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-        build_auth_headers(&url, &Method::GET, &mut headers)?;
-        Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?))
+        self.data_transport
+            .open_read(ReadStreamRequest {
+                endpoint: self.endpoint.grid_host(),
+                disk,
+                volume: volume.to_string(),
+                path: path.to_string(),
+                offset,
+                length,
+            })
+            .await
     }
 
     /// Zero-copy read for remote disks falls back to efficient network read.
@@ -1291,21 +1279,16 @@ impl DiskAPI for RemoteDisk {
             return Err(DiskError::FaultyDisk);
         }
         let disk = self.disk_ref().await;
-
-        let url = format!(
-            "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
-            self.endpoint.grid_host(),
-            urlencoding::encode(&disk),
-            urlencoding::encode(volume),
-            urlencoding::encode(path),
-            true,
-            0
-        );
-
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-        build_auth_headers(&url, &Method::PUT, &mut headers)?;
-        Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?))
+        self.data_transport
+            .open_write(WriteStreamRequest {
+                endpoint: self.endpoint.grid_host(),
+                disk,
+                volume: volume.to_string(),
+                path: path.to_string(),
+                append: true,
+                size: 0,
+            })
+            .await
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
@@ -1322,21 +1305,16 @@ impl DiskAPI for RemoteDisk {
             return Err(DiskError::FaultyDisk);
         }
         let disk = self.disk_ref().await;
-
-        let url = format!(
-            "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
-            self.endpoint.grid_host(),
-            urlencoding::encode(&disk),
-            urlencoding::encode(volume),
-            urlencoding::encode(path),
-            false,
-            file_size
-        );
-
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-        build_auth_headers(&url, &Method::PUT, &mut headers)?;
-        Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?))
+        self.data_transport
+            .open_write(WriteStreamRequest {
+                endpoint: self.endpoint.grid_host(),
+                disk,
+                volume: volume.to_string(),
+                path: path.to_string(),
+                append: false,
+                size: file_size,
+            })
+            .await
     }
 
     #[tracing::instrument(level = "debug", skip(self))]

scripts/run_internode_transport_baseline.sh

@@ -0,0 +1,327 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Internode transport baseline runner.
+# Reuses scripts/run_object_batch_bench.sh and exports reproducible artifacts:
+# - object benchmark summaries per scenario/workload/concurrency
+# - optional internode operation metric deltas from a Prometheus text endpoint
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+OBJECT_BENCH_SCRIPT="${PROJECT_ROOT}/scripts/run_object_batch_bench.sh"
+
+TOOL="warp"
+ACCESS_KEY=""
+SECRET_KEY=""
+REGION="us-east-1"
+BUCKET_PREFIX="rustfs-internode-bench"
+OUT_DIR=""
+SIZES="4KiB,1MiB,16MiB,128MiB,1GiB"
+CONCURRENCIES="1,16,64,128"
+DURATION="90s"
+SAMPLES=20000
+INSECURE=false
+EXTRA_ARGS=""
+WARP_BIN="warp"
+S3BENCH_BIN="s3bench"
+
+# Scenario format: name=endpoint
+SCENARIOS="local=http://127.0.0.1:9000,distributed=http://127.0.0.1:9001"
+
+# Optional Prometheus text exposition URL.
+# If empty, metrics delta collection is skipped.
+INTERNODE_METRICS_URL=""
+
+usage() {
+  cat <<'USAGE'
+Usage:
+  scripts/run_internode_transport_baseline.sh --access-key <ak> --secret-key <sk> [options]
+
+Required:
+  --access-key <ak>
+  --secret-key <sk>
+
+Optional:
+  --tool <warp|s3bench>              Default: warp
+  --region <name>                    Default: us-east-1
+  --bucket-prefix <prefix>           Default: rustfs-internode-bench
+  --scenarios <name=url,...>         Default: local=http://127.0.0.1:9000,distributed=http://127.0.0.1:9001
+  --sizes <csv>                      Default: 4KiB,1MiB,16MiB,128MiB,1GiB
+  --concurrencies <csv>              Default: 1,16,64,128
+  --duration <dur>                   Default: 90s
+  --samples <n>                      Default: 20000 (for s3bench)
+  --warp-bin <path>                  Default: warp
+  --s3bench-bin <path>               Default: s3bench
+  --metrics-url <url>                Prometheus text endpoint for internode metrics delta
+  --out-dir <dir>                    Default: target/bench/internode-transport-<timestamp>
+  --extra-args "<args>"              Passed to run_object_batch_bench.sh --extra-args
+  --insecure                         TLS insecure (self-signed)
+  --dry-run                          Print commands only
+  -h, --help
+
+Notes:
+  - This baseline covers S3 PUT/GET workloads and records internode metric deltas when --metrics-url is set.
+  - Healing/replication-specific workloads should be run separately and appended to the same artifact directory.
+USAGE
+}
+
+require_cmd() {
+  if ! command -v "$1" >/dev/null 2>&1; then
+    echo "ERROR: command not found: $1" >&2
+    exit 1
+  fi
+}
+
+DRY_RUN=false
+parse_args() {
+  while [[ $# -gt 0 ]]; do
+    case "$1" in
+      --access-key) ACCESS_KEY="$2"; shift 2 ;;
+      --secret-key) SECRET_KEY="$2"; shift 2 ;;
+      --tool) TOOL="$2"; shift 2 ;;
+      --region) REGION="$2"; shift 2 ;;
+      --bucket-prefix) BUCKET_PREFIX="$2"; shift 2 ;;
+      --scenarios) SCENARIOS="$2"; shift 2 ;;
+      --sizes) SIZES="$2"; shift 2 ;;
+      --concurrencies) CONCURRENCIES="$2"; shift 2 ;;
+      --duration) DURATION="$2"; shift 2 ;;
+      --samples) SAMPLES="$2"; shift 2 ;;
+      --warp-bin) WARP_BIN="$2"; shift 2 ;;
+      --s3bench-bin) S3BENCH_BIN="$2"; shift 2 ;;
+      --metrics-url) INTERNODE_METRICS_URL="$2"; shift 2 ;;
+      --out-dir) OUT_DIR="$2"; shift 2 ;;
+      --extra-args) EXTRA_ARGS="$2"; shift 2 ;;
+      --insecure) INSECURE=true; shift ;;
+      --dry-run) DRY_RUN=true; shift ;;
+      -h|--help) usage; exit 0 ;;
+      *)
+        echo "ERROR: unknown arg: $1" >&2
+        usage
+        exit 1
+        ;;
+    esac
+  done
+}
+
+validate_args() {
+  if [[ -z "${ACCESS_KEY}" || -z "${SECRET_KEY}" ]]; then
+    echo "ERROR: --access-key and --secret-key are required" >&2
+    exit 1
+  fi
+  if [[ "${TOOL}" != "warp" && "${TOOL}" != "s3bench" ]]; then
+    echo "ERROR: --tool must be warp or s3bench" >&2
+    exit 1
+  fi
+  if [[ -z "${SCENARIOS}" ]]; then
+    echo "ERROR: --scenarios cannot be empty" >&2
+    exit 1
+  fi
+}
+
+setup_output() {
+  if [[ -z "${OUT_DIR}" ]]; then
+    OUT_DIR="target/bench/internode-transport-$(date +%Y%m%d-%H%M%S)"
+  fi
+  mkdir -p "${OUT_DIR}"
+  echo "scenario,endpoint,workload,concurrency,size,status,throughput,requests_per_sec,avg_latency,log_file,run_dir" > "${OUT_DIR}/summary.csv"
+  if [[ -n "${INTERNODE_METRICS_URL}" ]]; then
+    echo "scenario,workload,concurrency,size,metric,operation,before,after,delta" > "${OUT_DIR}/internode_metric_deltas.csv"
+  fi
+}
+
+collect_internode_snapshot() {
+  local snapshot_file="$1"
+  if [[ -z "${INTERNODE_METRICS_URL}" ]]; then
+    : > "${snapshot_file}"
+    return 0
+  fi
+  if [[ "${DRY_RUN}" == "true" ]]; then
+    : > "${snapshot_file}"
+    return 0
+  fi
+  if ! curl -fsSL "${INTERNODE_METRICS_URL}" > "${snapshot_file}"; then
+    echo "WARN: failed to fetch metrics from ${INTERNODE_METRICS_URL}, skipping metrics delta for this run" >&2
+    : > "${snapshot_file}"
+  fi
+}
+
+extract_internode_rows() {
+  local src="$1"
+  if [[ ! -s "${src}" ]]; then
+    return 0
+  fi
+  awk '
+  $1 ~ /^rustfs_system_network_internode_operation_/ {
+    metric = $1
+    op = "all"
+    if (match($0, /operation="[^"]+"/)) {
+      op = substr($0, RSTART + 11, RLENGTH - 12)
+    }
+    n = split($0, parts, " ")
+    value = parts[n]
+    gsub(/[[:space:]]+/, "", value)
+    if (value ~ /^[0-9]+([.][0-9]+)?$/) {
+      print metric "," op "," value
+    }
+  }' "${src}"
+}
+
+append_metric_deltas() {
+  local scenario="$1"
+  local workload="$2"
+  local conc="$3"
+  local size="$4"
+  local before_file="$5"
+  local after_file="$6"
+
+  local before_rows after_rows
+  before_rows="$(mktemp)"
+  after_rows="$(mktemp)"
+  extract_internode_rows "${before_file}" > "${before_rows}"
+  extract_internode_rows "${after_file}" > "${after_rows}"
+
+  awk -F',' -v scenario="${scenario}" -v workload="${workload}" -v conc="${conc}" -v size="${size}" '
+    FNR==NR {
+      key = $1 SUBSEP $2
+      before[key] = $3
+      next
+    }
+    {
+      key = $1 SUBSEP $2
+      metric = $1
+      operation = $2
+      afterv = $3 + 0
+      beforev = (key in before ? before[key] + 0 : 0)
+      delta = afterv - beforev
+      printf "%s,%s,%s,%s,%s,%s,%.0f,%.0f,%.0f\n", scenario, workload, conc, size, metric, operation, beforev, afterv, delta
+    }
+  ' "${before_rows}" "${after_rows}" >> "${OUT_DIR}/internode_metric_deltas.csv"
+
+  rm -f "${before_rows}" "${after_rows}"
+}
+
+append_object_summary() {
+  local scenario="$1"
+  local endpoint="$2"
+  local workload="$3"
+  local conc="$4"
+  local run_dir="$5"
+
+  local src="${run_dir}/summary.csv"
+  if [[ ! -f "${src}" ]]; then
+    echo "WARN: missing summary file: ${src}" >&2
+    return 0
+  fi
+
+  awk -F',' -v scenario="${scenario}" -v endpoint="${endpoint}" -v workload="${workload}" -v run_dir="${run_dir}" '
+    NR == 1 { next }
+    {
+      printf "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n", scenario, endpoint, workload, $3, $1, $4, $5, $6, $7, $8, run_dir
+    }
+  ' "${src}" >> "${OUT_DIR}/summary.csv"
+}
+
+run_workload() {
+  local scenario="$1"
+  local endpoint="$2"
+  local workload="$3"
+  local conc="$4"
+
+  local run_dir="${OUT_DIR}/${scenario}/${workload}/concurrency-${conc}"
+  local bucket="${BUCKET_PREFIX}-${scenario}-${workload}-c${conc}"
+  local before_metrics="${run_dir}/metrics_before.prom"
+  local after_metrics="${run_dir}/metrics_after.prom"
+
+  mkdir -p "${run_dir}"
+  if [[ -n "${INTERNODE_METRICS_URL}" ]]; then
+    collect_internode_snapshot "${before_metrics}"
+  fi
+
+  local cmd=(
+    "${OBJECT_BENCH_SCRIPT}"
+    --tool "${TOOL}"
+    --endpoint "${endpoint}"
+    --access-key "${ACCESS_KEY}"
+    --secret-key "${SECRET_KEY}"
+    --bucket "${bucket}"
+    --region "${REGION}"
+    --concurrency "${conc}"
+    --sizes "${SIZES}"
+    --out-dir "${run_dir}"
+  )
+  if [[ "${TOOL}" == "warp" ]]; then
+    cmd+=(--duration "${DURATION}" --warp-mode "${workload}" --warp-bin "${WARP_BIN}")
+  else
+    cmd+=(--samples "${SAMPLES}" --s3bench-bin "${S3BENCH_BIN}")
+  fi
+  if [[ "${INSECURE}" == "true" ]]; then
+    cmd+=(--insecure)
+  fi
+  if [[ -n "${EXTRA_ARGS}" ]]; then
+    cmd+=(--extra-args "${EXTRA_ARGS}")
+  fi
+  if [[ "${DRY_RUN}" == "true" ]]; then
+    cmd+=(--dry-run)
+  fi
+
+  printf '==> scenario=%s workload=%s concurrency=%s endpoint=%s\n' "${scenario}" "${workload}" "${conc}" "${endpoint}"
+  "${cmd[@]}"
+
+  append_object_summary "${scenario}" "${endpoint}" "${workload}" "${conc}" "${run_dir}"
+  if [[ -n "${INTERNODE_METRICS_URL}" ]]; then
+    collect_internode_snapshot "${after_metrics}"
+    append_metric_deltas "${scenario}" "${workload}" "${conc}" "all_sizes" "${before_metrics}" "${after_metrics}"
+  fi
+}
+
+run_all() {
+  local scenario_pair scenario endpoint
+  local workload conc
+
+  IFS=',' read -r -a scenario_list <<< "${SCENARIOS}"
+  IFS=',' read -r -a conc_list <<< "${CONCURRENCIES}"
+
+  for scenario_pair in "${scenario_list[@]}"; do
+    scenario="${scenario_pair%%=*}"
+    endpoint="${scenario_pair#*=}"
+    if [[ -z "${scenario}" || -z "${endpoint}" || "${scenario}" == "${endpoint}" ]]; then
+      echo "ERROR: invalid scenario entry: ${scenario_pair} (expected name=url)" >&2
+      exit 1
+    fi
+
+    for workload in put get; do
+      for conc in "${conc_list[@]}"; do
+        conc="$(echo "${conc}" | xargs)"
+        if ! [[ "${conc}" =~ ^[0-9]+$ ]] || [[ "${conc}" -le 0 ]]; then
+          echo "ERROR: invalid concurrency: ${conc}" >&2
+          exit 1
+        fi
+        run_workload "${scenario}" "${endpoint}" "${workload}" "${conc}"
+      done
+    done
+  done
+}
+
+main() {
+  parse_args "$@"
+  validate_args
+  require_cmd awk
+  if [[ -n "${INTERNODE_METRICS_URL}" && "${DRY_RUN}" != "true" ]]; then
+    require_cmd curl
+  fi
+  if [[ ! -x "${OBJECT_BENCH_SCRIPT}" ]]; then
+    echo "ERROR: benchmark script not executable: ${OBJECT_BENCH_SCRIPT}" >&2
+    exit 1
+  fi
+
+  setup_output
+  run_all
+
+  echo "Artifacts:"
+  echo "  ${OUT_DIR}/summary.csv"
+  if [[ -n "${INTERNODE_METRICS_URL}" ]]; then
+    echo "  ${OUT_DIR}/internode_metric_deltas.csv"
+  fi
+}
+
+main "$@"