feat(internode): harden data transport baseline

crates/ecstore/src/rpc/internode_data_transport.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::disk::error::Result;
+use crate::disk::error::{Error, Result};
 use crate::disk::{FileReader, FileWriter};
 use crate::rpc::build_auth_headers;
 use async_trait::async_trait;
@@ -20,10 +20,38 @@ use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE};
 use rustfs_rio::{HttpReader, HttpWriter};
 use std::sync::Arc;
 use std::time::Duration;
-use tracing::warn;
 
 pub const ENV_RUSTFS_INTERNODE_DATA_TRANSPORT: &str = "RUSTFS_INTERNODE_DATA_TRANSPORT";
 pub const INTERNODE_DATA_TRANSPORT_TCP_HTTP: &str = "tcp-http";
+pub const INTERNODE_DATA_TRANSPORT_TCP: &str = "tcp";
+
+const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream";
+const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream";
+const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir";
+const CONTENT_TYPE_JSON: &str = "application/json";
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub struct InternodeDataTransportCapabilities {
+    pub stream_read: bool,
+    pub stream_write: bool,
+    pub walk_dir: bool,
+    pub registered_memory: bool,
+    pub scatter_gather: bool,
+    pub zero_copy_receive: bool,
+}
+
+impl InternodeDataTransportCapabilities {
+    pub const fn tcp_http() -> Self {
+        Self {
+            stream_read: true,
+            stream_write: true,
+            walk_dir: true,
+            registered_memory: false,
+            scatter_gather: false,
+            zero_copy_receive: false,
+        }
+    }
+}
 
 #[derive(Debug, Clone)]
 pub struct ReadStreamRequest {
@@ -59,6 +87,7 @@ pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
     async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter>;
     async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader>;
     fn name(&self) -> &'static str;
+    fn capabilities(&self) -> InternodeDataTransportCapabilities;
 }
 
 #[derive(Debug, Default)]
@@ -67,41 +96,22 @@ 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"));
+        let url = build_read_file_stream_url(&request);
+        let mut headers = json_headers();
         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"));
+        let url = build_put_file_stream_url(&request);
+        let mut headers = json_headers();
         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"));
+        let url = build_walk_dir_url(&request);
+        let mut headers = json_headers();
         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?,
@@ -111,23 +121,163 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
     fn name(&self) -> &'static str {
         INTERNODE_DATA_TRANSPORT_TCP_HTTP
     }
-}
 
-pub fn build_internode_data_transport_from_env() -> Arc<dyn InternodeDataTransport> {
-    match std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT) {
-        Ok(transport) => {
-            if transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP_HTTP) {
-                Arc::new(TcpHttpInternodeDataTransport)
-            } else {
-                warn!(
-                    env = ENV_RUSTFS_INTERNODE_DATA_TRANSPORT,
-                    requested = %transport,
-                    fallback = INTERNODE_DATA_TRANSPORT_TCP_HTTP,
-                    "unknown internode data transport, using default backend"
-                );
-                Arc::new(TcpHttpInternodeDataTransport)
-            }
-        }
-        Err(_) => Arc::new(TcpHttpInternodeDataTransport),
+    fn capabilities(&self) -> InternodeDataTransportCapabilities {
+        InternodeDataTransportCapabilities::tcp_http()
+    }
+}
+
+fn build_read_file_stream_url(request: &ReadStreamRequest) -> String {
+    format!(
+        "{}{}?disk={}&volume={}&path={}&offset={}&length={}",
+        request.endpoint,
+        READ_FILE_STREAM_PATH,
+        urlencoding::encode(&request.disk),
+        urlencoding::encode(&request.volume),
+        urlencoding::encode(&request.path),
+        request.offset,
+        request.length
+    )
+}
+
+fn build_put_file_stream_url(request: &WriteStreamRequest) -> String {
+    format!(
+        "{}{}?disk={}&volume={}&path={}&append={}&size={}",
+        request.endpoint,
+        PUT_FILE_STREAM_PATH,
+        urlencoding::encode(&request.disk),
+        urlencoding::encode(&request.volume),
+        urlencoding::encode(&request.path),
+        request.append,
+        request.size
+    )
+}
+
+fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String {
+    format!("{}{}?disk={}", request.endpoint, WALK_DIR_PATH, urlencoding::encode(&request.disk))
+}
+
+fn json_headers() -> HeaderMap {
+    let mut headers = HeaderMap::new();
+    headers.insert(CONTENT_TYPE, HeaderValue::from_static(CONTENT_TYPE_JSON));
+    headers
+}
+
+pub fn build_internode_data_transport(configured_transport: Option<&str>) -> Result<Arc<dyn InternodeDataTransport>> {
+    match configured_transport.map(str::trim).filter(|transport| !transport.is_empty()) {
+        None => Ok(Arc::new(TcpHttpInternodeDataTransport)),
+        Some(transport)
+            if transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP_HTTP)
+                || transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP) =>
+        {
+            Ok(Arc::new(TcpHttpInternodeDataTransport))
+        }
+        Some(transport) => Err(Error::other(format!(
+            "unsupported internode data transport {transport:?}; supported values: {INTERNODE_DATA_TRANSPORT_TCP_HTTP}, {INTERNODE_DATA_TRANSPORT_TCP}"
+        ))),
+    }
+}
+
+pub fn build_internode_data_transport_from_env() -> Result<Arc<dyn InternodeDataTransport>> {
+    build_internode_data_transport(std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok().as_deref())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn tcp_http_capabilities_are_behavior_preserving() {
+        let transport = TcpHttpInternodeDataTransport;
+
+        assert_eq!(transport.name(), INTERNODE_DATA_TRANSPORT_TCP_HTTP);
+        assert_eq!(
+            transport.capabilities(),
+            InternodeDataTransportCapabilities {
+                stream_read: true,
+                stream_write: true,
+                walk_dir: true,
+                registered_memory: false,
+                scatter_gather: false,
+                zero_copy_receive: false,
+            }
+        );
+    }
+
+    #[test]
+    fn read_file_stream_url_encodes_query_values() {
+        let url = build_read_file_stream_url(&ReadStreamRequest {
+            endpoint: "http://node1:9000".to_string(),
+            disk: "http://node1:9000/data/rustfs0".to_string(),
+            volume: ".rustfs.sys".to_string(),
+            path: "pool.bin/../part.1".to_string(),
+            offset: 7,
+            length: 11,
+        });
+
+        assert_eq!(
+            url,
+            "http://node1:9000/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0&volume=.rustfs.sys&path=pool.bin%2F..%2Fpart.1&offset=7&length=11"
+        );
+    }
+
+    #[test]
+    fn put_file_stream_url_encodes_query_values() {
+        let url = build_put_file_stream_url(&WriteStreamRequest {
+            endpoint: "http://node1:9000".to_string(),
+            disk: "http://node1:9000/data/rustfs0".to_string(),
+            volume: "bucket".to_string(),
+            path: "object/part.1".to_string(),
+            append: false,
+            size: 4096,
+        });
+
+        assert_eq!(
+            url,
+            "http://node1:9000/rustfs/rpc/put_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0&volume=bucket&path=object%2Fpart.1&append=false&size=4096"
+        );
+    }
+
+    #[test]
+    fn walk_dir_url_encodes_disk_ref() {
+        let url = build_walk_dir_url(&WalkDirStreamRequest {
+            endpoint: "http://node1:9000".to_string(),
+            disk: "http://node1:9000/data/rustfs0".to_string(),
+            body: Vec::new(),
+            stall_timeout: None,
+        });
+
+        assert_eq!(
+            url,
+            "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0"
+        );
+    }
+
+    #[test]
+    fn transport_config_defaults_to_tcp_http() {
+        let transport = build_internode_data_transport(None).unwrap();
+
+        assert_eq!(transport.name(), INTERNODE_DATA_TRANSPORT_TCP_HTTP);
+    }
+
+    #[test]
+    fn transport_config_accepts_tcp_aliases() {
+        for configured in [
+            INTERNODE_DATA_TRANSPORT_TCP_HTTP,
+            INTERNODE_DATA_TRANSPORT_TCP,
+            "TCP-HTTP",
+            "TCP",
+        ] {
+            let transport = build_internode_data_transport(Some(configured)).unwrap();
+
+            assert_eq!(transport.name(), INTERNODE_DATA_TRANSPORT_TCP_HTTP);
+        }
+    }
+
+    #[test]
+    fn transport_config_rejects_unknown_backend() {
+        let err = build_internode_data_transport(Some("rdma")).expect_err("unknown backend should fail closed");
+
+        assert!(err.to_string().contains("unsupported internode data transport"));
     }
 }

crates/ecstore/src/rpc/remote_disk.rs

@@ -129,7 +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(),
+            data_transport: build_internode_data_transport_from_env()?,
         };
         record_drive_runtime_state(ep, RuntimeDriveHealthState::Online);
 

scripts/run_internode_transport_baseline.sh

@@ -3,6 +3,7 @@ set -euo pipefail
 
 # Internode transport baseline runner.
 # Reuses scripts/run_object_batch_bench.sh and exports reproducible artifacts:
+# - run manifest with scenario/tool metadata and git revision
 # - object benchmark summaries per scenario/workload/concurrency
 # - optional internode operation metric deltas from a Prometheus text endpoint
 
@@ -61,6 +62,7 @@ Optional:
 
 Notes:
   - This baseline covers S3 PUT/GET workloads and records internode metric deltas when --metrics-url is set.
+  - The run manifest intentionally omits access keys, secret keys, and extra args to avoid writing credentials to artifacts.
   - Healing/replication-specific workloads should be run separately and appended to the same artifact directory.
 USAGE
 }
@@ -127,6 +129,41 @@ setup_output() {
   echo "scenario,workload,concurrency,size,metric,operation,before,after,delta" > "${OUT_DIR}/internode_metric_deltas.csv"
 }
 
+write_run_manifest() {
+  local manifest="${OUT_DIR}/run_manifest.txt"
+  local git_commit git_dirty rustc_version
+
+  git_commit="$(git -C "${PROJECT_ROOT}" rev-parse HEAD 2>/dev/null || echo "unknown")"
+  if [[ -n "$(git -C "${PROJECT_ROOT}" status --porcelain 2>/dev/null || true)" ]]; then
+    git_dirty="true"
+  else
+    git_dirty="false"
+  fi
+  rustc_version="$(rustc --version 2>/dev/null || echo "unknown")"
+
+  {
+    echo "created_utc=$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+    echo "git_commit=${git_commit}"
+    echo "git_dirty=${git_dirty}"
+    echo "rustc_version=${rustc_version}"
+    echo "kernel=$(uname -srvmo 2>/dev/null || echo "unknown")"
+    echo "tool=${TOOL}"
+    echo "region=${REGION}"
+    echo "bucket_prefix=${BUCKET_PREFIX}"
+    echo "scenarios=${SCENARIOS}"
+    echo "sizes=${SIZES}"
+    echo "concurrencies=${CONCURRENCIES}"
+    echo "duration=${DURATION}"
+    echo "samples=${SAMPLES}"
+    echo "insecure=${INSECURE}"
+    echo "metrics_url=${INTERNODE_METRICS_URL:-N/A}"
+    echo "out_dir=${OUT_DIR}"
+    echo "extra_args_present=$([[ -n "${EXTRA_ARGS}" ]] && echo true || echo false)"
+    echo "access_key=REDACTED"
+    echo "secret_key=REDACTED"
+  } > "${manifest}"
+}
+
 collect_internode_snapshot() {
   local snapshot_file="$1"
   if [[ -z "${INTERNODE_METRICS_URL}" ]]; then
@@ -300,17 +337,20 @@ main() {
   parse_args "$@"
   validate_args
   require_cmd awk
-  require_cmd curl
-  require_cmd rg
+  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
+  write_run_manifest
   run_all
 
   echo "Artifacts:"
+  echo "  ${OUT_DIR}/run_manifest.txt"
   echo "  ${OUT_DIR}/summary.csv"
   echo "  ${OUT_DIR}/internode_metric_deltas.csv"
 }

scripts/run_object_batch_bench.sh

@@ -74,6 +74,34 @@ require_cmd() {
   fi
 }
 
+print_dry_run_command() {
+  local redact_next=false
+  local arg
+
+  printf '[DRY-RUN]'
+  for arg in "$@"; do
+    if [[ "${redact_next}" == "true" ]]; then
+      printf ' %q' "REDACTED"
+      redact_next=false
+      continue
+    fi
+
+    case "${arg}" in
+      --access-key|--secret-key)
+        printf ' %q' "${arg}"
+        redact_next=true
+        ;;
+      -accessKey=*|-secretKey=*)
+        printf ' %q' "${arg%%=*}=REDACTED"
+        ;;
+      *)
+        printf ' %q' "${arg}"
+        ;;
+    esac
+  done
+  printf '\n'
+}
+
 normalize_warp_host() {
   local raw="$1"
   raw="${raw#http://}"
@@ -200,8 +228,7 @@ run_one() {
     fi
 
     if [[ "$DRY_RUN" == "true" ]]; then
-      printf '[DRY-RUN] %q ' "${cmd[@]}"
-      printf '\n'
+      print_dry_run_command "${cmd[@]}"
       echo "size=$size tool=$TOOL dry_run" > "$log_file"
     else
       if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then
@@ -228,8 +255,7 @@ run_one() {
     fi
 
     if [[ "$DRY_RUN" == "true" ]]; then
-      printf '[DRY-RUN] %q ' "${cmd[@]}"
-      printf '\n'
+      print_dry_run_command "${cmd[@]}"
       echo "size=$size tool=$TOOL dry_run" > "$log_file"
     else
       if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then
@@ -251,10 +277,12 @@ main() {
   parse_args "$@"
   validate_args
   require_cmd rg
-  if [[ "$TOOL" == "warp" ]]; then
-    require_cmd "$WARP_BIN"
-  else
-    require_cmd "$S3BENCH_BIN"
+  if [[ "$DRY_RUN" != "true" ]]; then
+    if [[ "$TOOL" == "warp" ]]; then
+      require_cmd "$WARP_BIN"
+    else
+      require_cmd "$S3BENCH_BIN"
+    fi
   fi
 
   setup_output