diff --git a/.docker/compat/README.md b/.docker/compat/README.md
new file mode 100644
index 000000000..b22747c33
--- /dev/null
+++ b/.docker/compat/README.md
@@ -0,0 +1,91 @@
+# Rio compatibility compose files
+
+These compose files prepare 4-node, 4-disk clusters for rio/rio-v2 storage format compatibility checks. All disks are bind-mounted under `.docker/compat/data` so the on-disk files remain available on the host.
+
+## Clusters
+
+```bash
+docker compose -f .docker/compat/docker-compose.rustfs-beta5.yml up -d --build
+docker compose -f .docker/compat/docker-compose.minio.yml up -d
+docker compose -f .docker/compat/docker-compose.rustfs-rio-v2.yml up -d --build
+```
+
+Default API endpoints:
+
+- RustFS `1.0.0-beta.5`: `http://127.0.0.1:9100`
+- MinIO: `http://127.0.0.1:9200`
+- current main with `rio-v2`: `http://127.0.0.1:9300`
+
+## Reading old datasets with rio-v2
+
+Stop the writer cluster before mounting its disks into the rio-v2 cluster.
+
+```bash
+docker compose -f .docker/compat/docker-compose.rustfs-beta5.yml down
+RUSTFS_RIO_V2_DATASET=./data/rustfs-beta5 \
+  docker compose -f .docker/compat/docker-compose.rustfs-rio-v2.yml up -d --build
+
+docker compose -f .docker/compat/docker-compose.minio.yml down
+RUSTFS_RIO_V2_DATASET=./data/minio \
+  docker compose -f .docker/compat/docker-compose.rustfs-rio-v2.yml up -d --build
+```
+
+## 200G object mix
+
+Use the same bucket/object matrix against the beta5 and MinIO endpoints, then read it back through the rio-v2 endpoint. A practical 200G mix is:
+
+- 1 KiB x 1024
+- 1 MiB x 1024
+- 64 MiB x 512
+- 1 GiB x 64
+- 8 GiB x 12
+- 6 GiB x 1
+
+Compression is enabled by default for RustFS and MinIO. Server-side KMS/SSE settings are intentionally left to environment variables or mounted key directories so real key material is not committed. For SSE-C cases, run the clusters with TLS because MinIO requires HTTPS for SSE-C.
+
+## High-concurrency write/read stress
+
+Use `run_rw_compat_stress.sh` to generate a manifest on an old endpoint, then verify the same objects through the rio-v2 endpoint after mounting the old disks.
+
+```bash
+.docker/compat/run_rw_compat_stress.sh \
+  --mode write \
+  --endpoint http://127.0.0.1:9100 \
+  --access-key rustfsadmin \
+  --secret-key rustfsadmin \
+  --bucket compat-beta5 \
+  --concurrency 96
+
+RUSTFS_RIO_V2_DATASET=./data/rustfs-beta5 \
+  docker compose -f .docker/compat/docker-compose.rustfs-rio-v2.yml up -d --build
+
+.docker/compat/run_rw_compat_stress.sh \
+  --mode verify \
+  --endpoint http://127.0.0.1:9300 \
+  --access-key rustfsadmin \
+  --secret-key rustfsadmin \
+  --bucket compat-beta5 \
+  --concurrency 96 \
+  --manifest target/compat/rw-stress-YYYYmmdd-HHMMSS/manifest.csv
+```
+
+For encrypted datasets, add `--encryption sse-s3`, `--encryption sse-kms --sse-kms-key-id <key-id>`, or `--encryption sse-c --sse-c-key-file <raw-32-byte-key-file>` to both the write and verify commands.
+
+## 5 GiB encrypted compatibility run
+
+The `5g` profile covers 1 KiB, 1 MiB, 16 MiB, 64 MiB, and 1 GiB objects and totals exactly 5 GiB. Generate `compat-key.key` under `.docker/compat/kms/rustfs-compat`, enable local KMS on both RustFS clusters, then use the same encryption arguments while writing with beta5 and verifying with rio-v2. Set non-default local test credentials first because distributed listeners reject the built-in default credentials.
+
+```bash
+export COMPAT_ACCESS_KEY='<non-default-access-key>'
+export COMPAT_SECRET_KEY='<non-default-secret-key>'
+
+RUSTFS_ACCESS_KEY="$COMPAT_ACCESS_KEY" RUSTFS_SECRET_KEY="$COMPAT_SECRET_KEY" \
+  RUSTFS_KMS_ENABLE=true RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS=true \
+  docker compose -f .docker/compat/docker-compose.rustfs-beta5.yml up -d
+
+.docker/compat/run_rw_compat_stress.sh \
+  --mode write --endpoint http://127.0.0.1:9100 \
+  --access-key "$COMPAT_ACCESS_KEY" --secret-key "$COMPAT_SECRET_KEY" \
+  --bucket compat-beta5-sse-s3 --profile 5g --concurrency 16 \
+  --encryption sse-s3
+```
diff --git a/.docker/compat/docker-compose.minio.yml b/.docker/compat/docker-compose.minio.yml
new file mode 100644
index 000000000..9ab79dee7
--- /dev/null
+++ b/.docker/compat/docker-compose.minio.yml
@@ -0,0 +1,88 @@
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0.
+
+x-minio-env: &minio-env
+  MINIO_ROOT_USER: "${MINIO_ROOT_USER:-minioadmin}"
+  MINIO_ROOT_PASSWORD: "${MINIO_ROOT_PASSWORD:-minioadmin}"
+  MINIO_COMPRESSION_ENABLE: "${MINIO_COMPRESSION_ENABLE:-on}"
+  MINIO_COMPRESSION_ALLOW_ENCRYPTION: "${MINIO_COMPRESSION_ALLOW_ENCRYPTION:-on}"
+  MINIO_COMPRESSION_EXTENSIONS: "${MINIO_COMPRESSION_EXTENSIONS:-.txt,.log,.csv,.json,.tar,.xml,.bin}"
+  MINIO_COMPRESSION_MIME_TYPES: "${MINIO_COMPRESSION_MIME_TYPES:-text/*,application/json,application/xml,binary/octet-stream}"
+
+x-minio-node: &minio-node
+  image: "${MINIO_IMAGE:-quay.io/minio/minio:latest}"
+  command: server --console-address ":9001" "http://minio{1...4}:9000/data/disk{1...4}"
+  environment: *minio-env
+  networks:
+    - minio-compat-net
+  restart: unless-stopped
+
+services:
+  minio-permission-helper:
+    image: alpine:3.23
+    command: sh -c "mkdir -p /compat-data && chown -R 1000:1000 /compat-data"
+    volumes:
+      - ./data/minio:/compat-data
+    restart: "no"
+
+  minio1:
+    <<: *minio-node
+    hostname: minio1
+    depends_on:
+      minio-permission-helper:
+        condition: service_completed_successfully
+    volumes:
+      - ./data/minio/node1/disk1:/data/disk1
+      - ./data/minio/node1/disk2:/data/disk2
+      - ./data/minio/node1/disk3:/data/disk3
+      - ./data/minio/node1/disk4:/data/disk4
+    ports:
+      - "${MINIO_API_PORT:-9200}:9000"
+      - "${MINIO_CONSOLE_PORT:-9201}:9001"
+
+  minio2:
+    <<: *minio-node
+    hostname: minio2
+    depends_on:
+      minio-permission-helper:
+        condition: service_completed_successfully
+    volumes:
+      - ./data/minio/node2/disk1:/data/disk1
+      - ./data/minio/node2/disk2:/data/disk2
+      - ./data/minio/node2/disk3:/data/disk3
+      - ./data/minio/node2/disk4:/data/disk4
+    ports:
+      - "${MINIO_NODE2_PORT:-9202}:9000"
+
+  minio3:
+    <<: *minio-node
+    hostname: minio3
+    depends_on:
+      minio-permission-helper:
+        condition: service_completed_successfully
+    volumes:
+      - ./data/minio/node3/disk1:/data/disk1
+      - ./data/minio/node3/disk2:/data/disk2
+      - ./data/minio/node3/disk3:/data/disk3
+      - ./data/minio/node3/disk4:/data/disk4
+    ports:
+      - "${MINIO_NODE3_PORT:-9203}:9000"
+
+  minio4:
+    <<: *minio-node
+    hostname: minio4
+    depends_on:
+      minio-permission-helper:
+        condition: service_completed_successfully
+    volumes:
+      - ./data/minio/node4/disk1:/data/disk1
+      - ./data/minio/node4/disk2:/data/disk2
+      - ./data/minio/node4/disk3:/data/disk3
+      - ./data/minio/node4/disk4:/data/disk4
+    ports:
+      - "${MINIO_NODE4_PORT:-9204}:9000"
+
+networks:
+  minio-compat-net:
+    driver: bridge
diff --git a/.docker/compat/docker-compose.rustfs-beta5.yml b/.docker/compat/docker-compose.rustfs-beta5.yml
new file mode 100644
index 000000000..1bda29e41
--- /dev/null
+++ b/.docker/compat/docker-compose.rustfs-beta5.yml
@@ -0,0 +1,104 @@
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0.
+
+x-rustfs-env: &rustfs-env
+  RUSTFS_VOLUMES: "http://rustfs-beta5-node{1...4}:9000/data/disk{1...4}"
+  RUSTFS_ADDRESS: ":9000"
+  RUSTFS_CONSOLE_ADDRESS: ":9001"
+  RUSTFS_CONSOLE_ENABLE: "true"
+  RUSTFS_CONSOLE_CORS_ALLOWED_ORIGINS: "*"
+  RUSTFS_ACCESS_KEY: "${RUSTFS_ACCESS_KEY:-rustfsadmin}"
+  RUSTFS_SECRET_KEY: "${RUSTFS_SECRET_KEY:-rustfsadmin}"
+  RUSTFS_OBS_LOGGER_LEVEL: "${RUSTFS_OBS_LOGGER_LEVEL:-info}"
+  RUSTFS_OBS_LOG_DIRECTORY: "/logs"
+  RUSTFS_COMPRESSION_ENABLED: "${RUSTFS_COMPRESSION_ENABLED:-true}"
+  RUSTFS_COMPRESSION_EXTENSIONS: "${RUSTFS_COMPRESSION_EXTENSIONS:-.txt,.log,.csv,.json,.tar,.xml,.bin}"
+  RUSTFS_COMPRESSION_MIME_TYPES: "${RUSTFS_COMPRESSION_MIME_TYPES:-text/*,application/json,application/xml,binary/octet-stream}"
+  RUSTFS_KMS_ENABLE: "${RUSTFS_KMS_ENABLE:-false}"
+  RUSTFS_KMS_BACKEND: "${RUSTFS_KMS_BACKEND:-local}"
+  RUSTFS_KMS_KEY_DIR: "${RUSTFS_KMS_KEY_DIR:-/kms}"
+  RUSTFS_KMS_DEFAULT_KEY_ID: "${RUSTFS_KMS_DEFAULT_KEY_ID:-compat-key}"
+  RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS: "${RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS:-false}"
+  RUSTFS_UNSAFE_BYPASS_DISK_CHECK: "${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true}"
+
+x-rustfs-node: &rustfs-node
+  image: "${RUSTFS_BETA5_IMAGE:-rustfs/rustfs:1.0.0-beta.5}"
+  build:
+    context: ../..
+    dockerfile: Dockerfile
+    args:
+      RELEASE: "1.0.0-beta.5"
+  environment: *rustfs-env
+  depends_on:
+    rustfs-beta5-permission-helper:
+      condition: service_completed_successfully
+  networks:
+    - rustfs-beta5-net
+  restart: unless-stopped
+
+services:
+  rustfs-beta5-permission-helper:
+    image: alpine:3.23
+    command: sh -c "mkdir -p /compat-data /kms && chown -R 10001:10001 /compat-data /kms"
+    volumes:
+      - ./data/rustfs-beta5:/compat-data
+      - ./kms/rustfs-compat:/kms
+    restart: "no"
+
+  rustfs-beta5-node1:
+    <<: *rustfs-node
+    hostname: rustfs-beta5-node1
+    volumes:
+      - ./data/rustfs-beta5/node1/disk1:/data/disk1
+      - ./data/rustfs-beta5/node1/disk2:/data/disk2
+      - ./data/rustfs-beta5/node1/disk3:/data/disk3
+      - ./data/rustfs-beta5/node1/disk4:/data/disk4
+      - ./data/rustfs-beta5/logs/node1:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_BETA5_API_PORT:-9100}:9000"
+      - "${RUSTFS_BETA5_CONSOLE_PORT:-9101}:9001"
+
+  rustfs-beta5-node2:
+    <<: *rustfs-node
+    hostname: rustfs-beta5-node2
+    volumes:
+      - ./data/rustfs-beta5/node2/disk1:/data/disk1
+      - ./data/rustfs-beta5/node2/disk2:/data/disk2
+      - ./data/rustfs-beta5/node2/disk3:/data/disk3
+      - ./data/rustfs-beta5/node2/disk4:/data/disk4
+      - ./data/rustfs-beta5/logs/node2:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_BETA5_NODE2_PORT:-9102}:9000"
+
+  rustfs-beta5-node3:
+    <<: *rustfs-node
+    hostname: rustfs-beta5-node3
+    volumes:
+      - ./data/rustfs-beta5/node3/disk1:/data/disk1
+      - ./data/rustfs-beta5/node3/disk2:/data/disk2
+      - ./data/rustfs-beta5/node3/disk3:/data/disk3
+      - ./data/rustfs-beta5/node3/disk4:/data/disk4
+      - ./data/rustfs-beta5/logs/node3:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_BETA5_NODE3_PORT:-9103}:9000"
+
+  rustfs-beta5-node4:
+    <<: *rustfs-node
+    hostname: rustfs-beta5-node4
+    volumes:
+      - ./data/rustfs-beta5/node4/disk1:/data/disk1
+      - ./data/rustfs-beta5/node4/disk2:/data/disk2
+      - ./data/rustfs-beta5/node4/disk3:/data/disk3
+      - ./data/rustfs-beta5/node4/disk4:/data/disk4
+      - ./data/rustfs-beta5/logs/node4:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_BETA5_NODE4_PORT:-9104}:9000"
+
+networks:
+  rustfs-beta5-net:
+    driver: bridge
diff --git a/.docker/compat/docker-compose.rustfs-rio-v2.yml b/.docker/compat/docker-compose.rustfs-rio-v2.yml
new file mode 100644
index 000000000..35401046f
--- /dev/null
+++ b/.docker/compat/docker-compose.rustfs-rio-v2.yml
@@ -0,0 +1,115 @@
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0.
+
+x-rustfs-rio-v2-env: &rustfs-rio-v2-env
+  RUSTFS_VOLUMES: "http://rustfs-rio-v2-node{1...4}:9000/data/disk{1...4}"
+  RUSTFS_ADDRESS: ":9000"
+  RUSTFS_CONSOLE_ADDRESS: ":9001"
+  RUSTFS_CONSOLE_ENABLE: "true"
+  RUSTFS_CONSOLE_CORS_ALLOWED_ORIGINS: "*"
+  RUSTFS_ACCESS_KEY: "${RUSTFS_ACCESS_KEY:-rustfsadmin}"
+  RUSTFS_SECRET_KEY: "${RUSTFS_SECRET_KEY:-rustfsadmin}"
+  RUSTFS_OBS_LOGGER_LEVEL: "${RUSTFS_OBS_LOGGER_LEVEL:-info}"
+  RUSTFS_OBS_LOG_DIRECTORY: "/logs"
+  RUSTFS_COMPRESSION_ENABLED: "${RUSTFS_COMPRESSION_ENABLED:-true}"
+  RUSTFS_COMPRESSION_EXTENSIONS: "${RUSTFS_COMPRESSION_EXTENSIONS:-.txt,.log,.csv,.json,.tar,.xml,.bin}"
+  RUSTFS_COMPRESSION_MIME_TYPES: "${RUSTFS_COMPRESSION_MIME_TYPES:-text/*,application/json,application/xml,binary/octet-stream}"
+  RUSTFS_KMS_ENABLE: "${RUSTFS_KMS_ENABLE:-false}"
+  RUSTFS_KMS_BACKEND: "${RUSTFS_KMS_BACKEND:-local}"
+  RUSTFS_KMS_KEY_DIR: "${RUSTFS_KMS_KEY_DIR:-/kms}"
+  RUSTFS_KMS_DEFAULT_KEY_ID: "${RUSTFS_KMS_DEFAULT_KEY_ID:-compat-key}"
+  RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS: "${RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS:-false}"
+  RUSTFS_UNSAFE_BYPASS_DISK_CHECK: "${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true}"
+
+x-rustfs-rio-v2-node: &rustfs-rio-v2-node
+  image: "${RUSTFS_RIO_V2_IMAGE:-rustfs/rustfs:compat-rio-v2}"
+  entrypoint:
+    - /bin/sh
+    - -c
+    - |
+      until getent hosts rustfs-rio-v2-node1 >/dev/null &&
+        getent hosts rustfs-rio-v2-node2 >/dev/null &&
+        getent hosts rustfs-rio-v2-node3 >/dev/null &&
+        getent hosts rustfs-rio-v2-node4 >/dev/null; do
+        sleep 1
+      done
+      exec /entrypoint.sh /usr/bin/rustfs
+  build:
+    context: ../..
+    dockerfile: Dockerfile.source
+    args:
+      RUSTFS_BUILD_FEATURES: "rio-v2"
+  environment: *rustfs-rio-v2-env
+  depends_on:
+    rustfs-rio-v2-permission-helper:
+      condition: service_completed_successfully
+  networks:
+    - rustfs-rio-v2-net
+  restart: unless-stopped
+
+services:
+  rustfs-rio-v2-permission-helper:
+    image: alpine:3.23
+    command: sh -c "mkdir -p /compat-data /kms && chown -R 10001:10001 /compat-data /kms"
+    volumes:
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}:/compat-data
+      - ./kms/rustfs-compat:/kms
+    restart: "no"
+
+  rustfs-rio-v2-node1:
+    <<: *rustfs-rio-v2-node
+    hostname: rustfs-rio-v2-node1
+    volumes:
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node1/disk1:/data/disk1
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node1/disk2:/data/disk2
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node1/disk3:/data/disk3
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node1/disk4:/data/disk4
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/logs/node1:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_RIO_V2_API_PORT:-9300}:9000"
+      - "${RUSTFS_RIO_V2_CONSOLE_PORT:-9301}:9001"
+
+  rustfs-rio-v2-node2:
+    <<: *rustfs-rio-v2-node
+    hostname: rustfs-rio-v2-node2
+    volumes:
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node2/disk1:/data/disk1
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node2/disk2:/data/disk2
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node2/disk3:/data/disk3
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node2/disk4:/data/disk4
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/logs/node2:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_RIO_V2_NODE2_PORT:-9302}:9000"
+
+  rustfs-rio-v2-node3:
+    <<: *rustfs-rio-v2-node
+    hostname: rustfs-rio-v2-node3
+    volumes:
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node3/disk1:/data/disk1
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node3/disk2:/data/disk2
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node3/disk3:/data/disk3
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node3/disk4:/data/disk4
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/logs/node3:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_RIO_V2_NODE3_PORT:-9303}:9000"
+
+  rustfs-rio-v2-node4:
+    <<: *rustfs-rio-v2-node
+    hostname: rustfs-rio-v2-node4
+    volumes:
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node4/disk1:/data/disk1
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node4/disk2:/data/disk2
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node4/disk3:/data/disk3
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/node4/disk4:/data/disk4
+      - ${RUSTFS_RIO_V2_DATASET:-./data/rustfs-rio-v2}/logs/node4:/logs
+      - ./kms/rustfs-compat:/kms
+    ports:
+      - "${RUSTFS_RIO_V2_NODE4_PORT:-9304}:9000"
+
+networks:
+  rustfs-rio-v2-net:
+    driver: bridge
diff --git a/.docker/compat/run_rw_compat_stress.sh b/.docker/compat/run_rw_compat_stress.sh
new file mode 100755
index 000000000..44a21be0f
--- /dev/null
+++ b/.docker/compat/run_rw_compat_stress.sh
@@ -0,0 +1,639 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# High-concurrency S3 read/write stress runner for rio/rio-v2 format compatibility.
+# Write a manifest on an old endpoint, then verify the same manifest through rio-v2.
+
+MODE="write"
+ENDPOINT=""
+ACCESS_KEY="${AWS_ACCESS_KEY_ID:-}"
+SECRET_KEY="${AWS_SECRET_ACCESS_KEY:-}"
+BUCKET="compat-rw-stress"
+REGION="us-east-1"
+CONCURRENCY=64
+OUT_DIR=""
+WORK_DIR=""
+MANIFEST=""
+PROFILE="200g"
+OBJECT_SPEC=""
+DATA_PATTERN="compressible"
+ENCRYPTION="none"
+SSE_KMS_KEY_ID=""
+SSE_C_KEY_FILE=""
+CLIENT="mc"
+AWS_BIN="${AWS_BIN:-aws}"
+MC_BIN="${MC_BIN:-}"
+KEEP_PAYLOADS=false
+DRY_RUN=false
+RESUME=false
+
+usage() {
+  cat <<'USAGE'
+Usage:
+  .docker/compat/run_rw_compat_stress.sh --mode <write|verify|mixed> \
+    --endpoint <url> --access-key <ak> --secret-key <sk> [options]
+
+Modes:
+  write     Create bucket, upload objects concurrently, and write manifest.csv.
+  verify    Read objects concurrently from --endpoint and verify against manifest.csv.
+  mixed     Write and verify against the same endpoint.
+
+Required:
+  --endpoint              S3 endpoint URL, for example http://127.0.0.1:9100
+  --access-key            S3 access key
+  --secret-key            S3 secret key
+
+Core options:
+  --bucket                Bucket name (default: compat-rw-stress)
+  --region                Region (default: us-east-1)
+  --concurrency           Parallel object operations (default: 64)
+  --out-dir               Output directory (default: target/compat/rw-stress-<timestamp>)
+  --work-dir              Payload scratch directory (default: <out-dir>/payloads)
+  --manifest              Manifest path (default: <out-dir>/manifest.csv for write/mixed)
+  --profile               compact | 5g | 200g (default: 200g)
+  --object-spec           Override profile. Format: size:count,size:count
+                         Example: 1KiB:1024,1MiB:1024,64MiB:512,1GiB:64
+  --data-pattern          compressible | random | mixed (default: compressible)
+  --keep-payloads         Do not delete local payload files after upload
+  --resume                Skip write tasks that already have rows in <out-dir>/tasks/write-rows
+  --dry-run               Print planned tasks and commands without executing S3 operations
+  --client                mc | aws (default: mc)
+  --mc-bin                Path to mc binary (default: first tmp/mc.* or mc in PATH)
+  --aws-bin               Path to aws binary (used with --client aws)
+
+Encryption options:
+  --encryption            none | sse-s3 | sse-kms | sse-c (default: none)
+  --sse-kms-key-id        KMS key id for --encryption sse-kms
+  --sse-c-key-file        Raw 32-byte SSE-C key file for --encryption sse-c
+
+Examples:
+  # Generate the old RustFS beta5 dataset.
+  .docker/compat/run_rw_compat_stress.sh \
+    --mode write --endpoint http://127.0.0.1:9100 \
+    --access-key rustfsadmin --secret-key rustfsadmin \
+    --bucket compat-beta5 --concurrency 96
+
+  # Verify that dataset after mounting beta5 disks into the rio-v2 compose.
+  .docker/compat/run_rw_compat_stress.sh \
+    --mode verify --endpoint http://127.0.0.1:9300 \
+    --access-key rustfsadmin --secret-key rustfsadmin \
+    --bucket compat-beta5 --concurrency 96 \
+    --manifest target/compat/rw-stress-YYYYmmdd-HHMMSS/manifest.csv
+
+  # Faster smoke run.
+  .docker/compat/run_rw_compat_stress.sh \
+    --mode mixed --endpoint http://127.0.0.1:9300 \
+    --access-key rustfsadmin --secret-key rustfsadmin \
+    --profile compact --concurrency 16
+USAGE
+}
+
+die() {
+  echo "ERROR: $*" >&2
+  exit 1
+}
+
+require_cmd() {
+  command -v "$1" >/dev/null 2>&1 || die "command not found: $1"
+}
+
+parse_args() {
+  while [[ $# -gt 0 ]]; do
+    case "$1" in
+      --mode) MODE="$2"; shift 2 ;;
+      --endpoint) ENDPOINT="$2"; shift 2 ;;
+      --access-key) ACCESS_KEY="$2"; shift 2 ;;
+      --secret-key) SECRET_KEY="$2"; shift 2 ;;
+      --bucket) BUCKET="$2"; shift 2 ;;
+      --region) REGION="$2"; shift 2 ;;
+      --concurrency) CONCURRENCY="$2"; shift 2 ;;
+      --out-dir) OUT_DIR="$2"; shift 2 ;;
+      --work-dir) WORK_DIR="$2"; shift 2 ;;
+      --manifest) MANIFEST="$2"; shift 2 ;;
+      --profile) PROFILE="$2"; shift 2 ;;
+      --object-spec) OBJECT_SPEC="$2"; shift 2 ;;
+      --data-pattern) DATA_PATTERN="$2"; shift 2 ;;
+      --encryption) ENCRYPTION="$2"; shift 2 ;;
+      --sse-kms-key-id) SSE_KMS_KEY_ID="$2"; shift 2 ;;
+      --sse-c-key-file) SSE_C_KEY_FILE="$2"; shift 2 ;;
+      --client) CLIENT="$2"; shift 2 ;;
+      --mc-bin) MC_BIN="$2"; shift 2 ;;
+      --aws-bin) AWS_BIN="$2"; shift 2 ;;
+      --keep-payloads) KEEP_PAYLOADS=true; shift ;;
+      --resume) RESUME=true; shift ;;
+      --dry-run) DRY_RUN=true; shift ;;
+      -h|--help) usage; exit 0 ;;
+      *) die "unknown arg: $1" ;;
+    esac
+  done
+}
+
+validate_args() {
+  [[ "$MODE" =~ ^(write|verify|mixed)$ ]] || die "--mode must be write, verify, or mixed"
+  [[ "$CLIENT" =~ ^(mc|aws)$ ]] || die "--client must be mc or aws"
+  [[ "$PROFILE" =~ ^(compact|5g|200g)$ ]] || die "--profile must be compact, 5g, or 200g"
+  [[ "$DATA_PATTERN" =~ ^(compressible|random|mixed)$ ]] || die "--data-pattern must be compressible, random, or mixed"
+  [[ "$ENCRYPTION" =~ ^(none|sse-s3|sse-kms|sse-c)$ ]] || die "--encryption must be none, sse-s3, sse-kms, or sse-c"
+  [[ -n "$ENDPOINT" && -n "$ACCESS_KEY" && -n "$SECRET_KEY" ]] || die "--endpoint/--access-key/--secret-key are required"
+  [[ "$CONCURRENCY" =~ ^[0-9]+$ && "$CONCURRENCY" -gt 0 ]] || die "--concurrency must be a positive integer"
+  if [[ "$ENCRYPTION" == "sse-kms" && -z "$SSE_KMS_KEY_ID" ]]; then
+    die "--sse-kms-key-id is required for --encryption sse-kms"
+  fi
+  if [[ "$ENCRYPTION" == "sse-c" ]]; then
+    [[ -n "$SSE_C_KEY_FILE" && -f "$SSE_C_KEY_FILE" ]] || die "--sse-c-key-file must point to an existing key file"
+  fi
+  if [[ "$MODE" == "verify" && -z "$MANIFEST" ]]; then
+    die "--manifest is required for --mode verify"
+  fi
+}
+
+setup_paths() {
+  if [[ -z "$OUT_DIR" ]]; then
+    OUT_DIR="target/compat/rw-stress-$(date +%Y%m%d-%H%M%S)"
+  fi
+  if [[ -z "$WORK_DIR" ]]; then
+    WORK_DIR="$OUT_DIR/payloads"
+  fi
+  if [[ -z "$MANIFEST" ]]; then
+    MANIFEST="$OUT_DIR/manifest.csv"
+  fi
+
+  mkdir -p "$OUT_DIR" "$WORK_DIR" "$OUT_DIR/tasks" "$OUT_DIR/logs"
+  TASKS_FILE="$OUT_DIR/tasks/tasks.tsv"
+  WRITE_ROWS_DIR="$OUT_DIR/tasks/write-rows"
+  VERIFY_ROWS_DIR="$OUT_DIR/tasks/verify-rows"
+  MC_CONFIG_DIR_LOCAL="$OUT_DIR/mc-config"
+  MC_ALIAS="compat"
+  mkdir -p "$WRITE_ROWS_DIR" "$VERIFY_ROWS_DIR"
+}
+
+resolve_mc_bin() {
+  if [[ -n "$MC_BIN" ]]; then
+    echo "$MC_BIN"
+    return
+  fi
+
+  local candidate
+  candidate="$(find tmp -maxdepth 1 -type f -name 'mc.*' -perm -111 2>/dev/null | sort | tail -n 1 || true)"
+  if [[ -n "$candidate" ]]; then
+    echo "$candidate"
+    return
+  fi
+
+  command -v mc 2>/dev/null || true
+}
+
+size_to_bytes() {
+  local raw="$1"
+  local num unit
+  if [[ "$raw" =~ ^([0-9]+)(B|KiB|MiB|GiB|KB|MB|GB)?$ ]]; then
+    num="${BASH_REMATCH[1]}"
+    unit="${BASH_REMATCH[2]:-B}"
+  else
+    die "invalid size: $raw"
+  fi
+
+  case "$unit" in
+    B) echo "$num" ;;
+    KiB) echo $((num * 1024)) ;;
+    MiB) echo $((num * 1024 * 1024)) ;;
+    GiB) echo $((num * 1024 * 1024 * 1024)) ;;
+    KB) echo $((num * 1000)) ;;
+    MB) echo $((num * 1000 * 1000)) ;;
+    GB) echo $((num * 1000 * 1000 * 1000)) ;;
+    *) die "invalid size unit: $unit" ;;
+  esac
+}
+
+profile_spec() {
+  if [[ -n "$OBJECT_SPEC" ]]; then
+    echo "$OBJECT_SPEC"
+    return
+  fi
+
+  case "$PROFILE" in
+    compact)
+      echo "1KiB:64,1MiB:64,16MiB:16,128MiB:4"
+      ;;
+    5g)
+      echo "1KiB:1024,1MiB:255,16MiB:64,64MiB:28,1GiB:2"
+      ;;
+    200g)
+      echo "1KiB:1024,1MiB:1024,64MiB:512,1GiB:64,8GiB:12,6GiB:1"
+      ;;
+  esac
+}
+
+content_for_index() {
+  local index="$1"
+  case $((index % 3)) in
+    0) echo "txt|text/plain" ;;
+    1) echo "json|application/json" ;;
+    2) echo "bin|binary/octet-stream" ;;
+  esac
+}
+
+generate_tasks() {
+  local spec item size count bytes i content ext mime key seed index=0
+  spec="$(profile_spec)"
+  : > "$TASKS_FILE"
+
+  IFS=',' read -r -a items <<< "$spec"
+  for item in "${items[@]}"; do
+    item="${item//[[:space:]]/}"
+    [[ -n "$item" ]] || continue
+    [[ "$item" =~ ^([^:]+):([0-9]+)$ ]] || die "invalid object spec item: $item"
+    size="${BASH_REMATCH[1]}"
+    count="${BASH_REMATCH[2]}"
+    bytes="$(size_to_bytes "$size")"
+
+    for ((i = 1; i <= count; i++)); do
+      content="$(content_for_index "$index")"
+      ext="${content%%|*}"
+      mime="${content#*|}"
+      key="rw-stress/${size}/obj-$(printf '%06d' "$i").${ext}"
+      seed="${BUCKET}:${key}:${bytes}"
+      printf '%s\t%s\t%s\t%s\t%s\n' "$key" "$size" "$bytes" "$mime" "$seed" >> "$TASKS_FILE"
+      index=$((index + 1))
+    done
+  done
+}
+
+write_row_file_for_key() {
+  local key="$1"
+  echo "$WRITE_ROWS_DIR/${key//\//_}.csv"
+}
+
+prepare_write_input() {
+  WRITE_INPUT="$TASKS_FILE"
+  if [[ "$RESUME" != "true" ]]; then
+    return
+  fi
+
+  WRITE_INPUT="$OUT_DIR/tasks/write-input.tsv"
+  : > "$WRITE_INPUT"
+
+  local line key _size _bytes _mime _seed row_file
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    IFS=$'\t' read -r key _size _bytes _mime _seed <<< "$line"
+    row_file="$(write_row_file_for_key "$key")"
+    [[ -s "$row_file" ]] && continue
+    printf '%s\n' "$line" >> "$WRITE_INPUT"
+  done < "$TASKS_FILE"
+}
+
+print_plan() {
+  local total_objects total_bytes profile_label
+  if [[ -f "$TASKS_FILE" ]]; then
+    total_objects="$(wc -l < "$TASKS_FILE" | tr -d ' ')"
+    total_bytes="$(awk -F '\t' '{sum += $3} END {print sum + 0}' "$TASKS_FILE")"
+    profile_label="$(profile_spec)"
+  else
+    total_objects="$(awk 'END {count = NR - 1; if (count < 0) count = 0; print count}' "$MANIFEST")"
+    total_bytes="$(awk -F ',' 'NR > 1 {sum += $3} END {print sum + 0}' "$MANIFEST")"
+    profile_label="from manifest"
+  fi
+
+  cat <<PLAN
+Mode:        $MODE
+Endpoint:    $ENDPOINT
+Bucket:      $BUCKET
+Profile:     $profile_label
+Objects:     $total_objects
+Bytes:       $total_bytes
+Concurrency: $CONCURRENCY
+Encryption:  $ENCRYPTION
+Client:      $CLIENT
+Manifest:    $MANIFEST
+Out dir:     $OUT_DIR
+Work dir:    $WORK_DIR
+PLAN
+}
+
+aws_base() {
+  AWS_ACCESS_KEY_ID="$ACCESS_KEY" AWS_SECRET_ACCESS_KEY="$SECRET_KEY" AWS_DEFAULT_REGION="$REGION" \
+    "$AWS_BIN" --endpoint-url "$ENDPOINT" "$@"
+}
+
+mc_base() {
+  "$MC_BIN" --config-dir "$MC_CONFIG_DIR_LOCAL" "$@"
+}
+
+aws_cp_args() {
+  case "$ENCRYPTION" in
+    none) ;;
+    sse-s3) printf '%s\n' "--sse" "AES256" ;;
+    sse-kms) printf '%s\n' "--sse" "aws:kms" "--sse-kms-key-id" "$SSE_KMS_KEY_ID" ;;
+    sse-c) printf '%s\n' "--sse-c" "AES256" "--sse-c-key" "fileb://$SSE_C_KEY_FILE" ;;
+  esac
+}
+
+mc_enc_target() {
+  printf '%s/%s/rw-stress/' "$MC_ALIAS" "$BUCKET"
+}
+
+mc_cp_args() {
+  local target
+  target="$(mc_enc_target)"
+  case "$ENCRYPTION" in
+    none) ;;
+    sse-s3) printf '%s\n' "--enc-s3" "$target" ;;
+    sse-kms) printf '%s\n' "--enc-kms" "${target}=${SSE_KMS_KEY_ID}" ;;
+    sse-c)
+      local key_b64
+      key_b64="$(base64 < "$SSE_C_KEY_FILE" | tr -d '\n')"
+      printf '%s\n' "--enc-c" "${target}=${key_b64}"
+      ;;
+  esac
+}
+
+aws_get_args() {
+  if [[ "$ENCRYPTION" == "sse-c" ]]; then
+    printf '%s\n' "--sse-c" "AES256" "--sse-c-key" "fileb://$SSE_C_KEY_FILE"
+  fi
+}
+
+mc_get_args() {
+  if [[ "$ENCRYPTION" == "sse-c" ]]; then
+    local target key_b64
+    target="$(mc_enc_target)"
+    key_b64="$(base64 < "$SSE_C_KEY_FILE" | tr -d '\n')"
+    printf '%s\n' "--enc-c" "${target}=${key_b64}"
+  fi
+}
+
+setup_mc_alias() {
+  if [[ "$CLIENT" != "mc" || "$DRY_RUN" == "true" ]]; then
+    return
+  fi
+  mkdir -p "$MC_CONFIG_DIR_LOCAL"
+  mc_base alias set "$MC_ALIAS" "$ENDPOINT" "$ACCESS_KEY" "$SECRET_KEY" --api S3v4 --path auto >/dev/null
+}
+
+create_bucket_if_needed() {
+  if [[ "$DRY_RUN" == "true" ]]; then
+    echo "[DRY-RUN] create bucket if missing: $BUCKET"
+    return
+  fi
+
+  if [[ "$CLIENT" == "mc" ]]; then
+    mc_base mb --ignore-existing --region "$REGION" "$MC_ALIAS/$BUCKET" >/dev/null
+    return
+  fi
+
+  if aws_base s3api head-bucket --bucket "$BUCKET" >/dev/null 2>&1; then
+    return
+  fi
+
+  aws_base s3api create-bucket --bucket "$BUCKET" >/dev/null
+}
+
+payload_path_for_key() {
+  local key="$1"
+  echo "$WORK_DIR/${key//\//_}"
+}
+
+generate_payload() {
+  local file="$1"
+  local bytes="$2"
+  local seed="$3"
+  local pattern="$DATA_PATTERN"
+  mkdir -p "$(dirname "$file")"
+
+  if [[ "$pattern" == "mixed" ]]; then
+    if [[ $((bytes % 2)) -eq 0 ]]; then
+      pattern="compressible"
+    else
+      pattern="random"
+    fi
+  fi
+
+  if [[ "$pattern" == "random" ]]; then
+    head -c "$bytes" /dev/zero | openssl enc -aes-256-ctr -nosalt -pass "pass:$seed" -out "$file"
+  else
+    yes "$seed payload-for-rio-compatibility" | tr '\n' ' ' | head -c "$bytes" > "$file"
+  fi
+}
+
+sha256_file() {
+  shasum -a 256 "$1" | awk '{print $1}'
+}
+
+sha256_object() {
+  local key="$1"
+  shift
+  if [[ "$CLIENT" == "mc" ]]; then
+    mc_base cat "$@" "$MC_ALIAS/$BUCKET/$key" | shasum -a 256 | awk '{print $1}'
+  else
+    aws_base s3 cp "s3://$BUCKET/$key" - --no-progress "$@" | shasum -a 256 | awk '{print $1}'
+  fi
+}
+
+collect_args() {
+  local generator="$1"
+  extra_args=()
+  while IFS= read -r arg; do
+    extra_args+=("$arg")
+  done < <("$generator")
+}
+
+write_one() {
+  local line="$1"
+  local key size bytes mime seed file sha row_file
+  local -a extra_args
+  IFS=$'\t' read -r key size bytes mime seed <<< "$line"
+  file="$(payload_path_for_key "$key")"
+  row_file="$(write_row_file_for_key "$key")"
+
+  if [[ "$DRY_RUN" == "true" ]]; then
+    echo "[DRY-RUN] upload $bytes bytes to s3://$BUCKET/$key content-type=$mime"
+    printf '%s,%s,%s,%s,%s,%s\n' "$key" "$size" "$bytes" "$mime" "DRY_RUN" "$ENCRYPTION" > "$row_file"
+    return
+  fi
+
+  generate_payload "$file" "$bytes" "$seed"
+  sha="$(sha256_file "$file")"
+
+  if [[ "$CLIENT" == "mc" ]]; then
+    collect_args mc_cp_args
+    mc_base cp --quiet --attr "Content-Type=$mime" ${extra_args[@]+"${extra_args[@]}"} "$file" "$MC_ALIAS/$BUCKET/$key" \
+      > "$OUT_DIR/logs/${key//\//_}.put.log" 2>&1
+  else
+    collect_args aws_cp_args
+    aws_base s3 cp "$file" "s3://$BUCKET/$key" --no-progress --content-type "$mime" ${extra_args[@]+"${extra_args[@]}"} \
+      > "$OUT_DIR/logs/${key//\//_}.put.log" 2>&1
+  fi
+
+  printf '%s,%s,%s,%s,%s,%s\n' "$key" "$size" "$bytes" "$mime" "$sha" "$ENCRYPTION" > "$row_file"
+  if [[ "$KEEP_PAYLOADS" != "true" ]]; then
+    rm -f "$file"
+  fi
+}
+
+verify_one() {
+  local line="$1"
+  local key size bytes mime expected encryption actual row_file
+  local -a extra_args
+  IFS=',' read -r key size bytes mime expected encryption <<< "$line"
+  row_file="$VERIFY_ROWS_DIR/${key//\//_}.csv"
+
+  if [[ "$DRY_RUN" == "true" ]]; then
+    echo "[DRY-RUN] verify s3://$BUCKET/$key expected=$expected"
+    printf '%s,%s,%s,%s,%s,%s,%s\n' "$key" "$size" "$bytes" "$mime" "$expected" "DRY_RUN" "dry-run" > "$row_file"
+    return
+  fi
+
+  if [[ "$CLIENT" == "mc" ]]; then
+    collect_args mc_get_args
+  else
+    collect_args aws_get_args
+  fi
+  if ! actual="$(sha256_object "$key" ${extra_args[@]+"${extra_args[@]}"})"; then
+    printf '%s,%s,%s,%s,%s,%s,%s\n' "$key" "$size" "$bytes" "$mime" "$expected" "ERROR" "download failed" > "$row_file"
+    return 1
+  fi
+
+  if [[ "$actual" == "$expected" ]]; then
+    printf '%s,%s,%s,%s,%s,%s,%s\n' "$key" "$size" "$bytes" "$mime" "$expected" "$actual" "ok" > "$row_file"
+  else
+    printf '%s,%s,%s,%s,%s,%s,%s\n' "$key" "$size" "$bytes" "$mime" "$expected" "$actual" "sha256-mismatch" > "$row_file"
+    return 1
+  fi
+}
+
+run_parallel_tasks() {
+  local action="$1"
+  local input_file="$2"
+  local failure_file="$OUT_DIR/tasks/${action}.failed"
+  local line
+  rm -f "$failure_file"
+
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    while [[ "$(jobs -rp | wc -l | tr -d ' ')" -ge "$CONCURRENCY" ]]; do
+      sleep 0.2
+    done
+
+    ({
+      if [[ "$action" == "write" ]]; then
+        write_one "$line"
+      else
+        verify_one "$line"
+      fi
+    } || touch "$failure_file") &
+  done < "$input_file"
+
+  wait
+  [[ ! -f "$failure_file" ]]
+}
+
+combine_write_manifest() {
+  echo "key,size,bytes,content_type,sha256,encryption" > "$MANIFEST"
+  find "$WRITE_ROWS_DIR" -type f -name '*.csv' -print0 | sort -z | xargs -0 cat >> "$MANIFEST"
+
+  local expected actual
+  expected="$(wc -l < "$TASKS_FILE" | tr -d ' ')"
+  actual="$(( $(wc -l < "$MANIFEST" | tr -d ' ') - 1 ))"
+  if [[ "$actual" -ne "$expected" ]]; then
+    die "manifest row count mismatch: expected $expected, got $actual"
+  fi
+}
+
+prepare_verify_input() {
+  VERIFY_INPUT="$OUT_DIR/tasks/verify-input.csv"
+  tail -n +2 "$MANIFEST" > "$VERIFY_INPUT"
+}
+
+combine_verify_summary() {
+  VERIFY_SUMMARY="$OUT_DIR/verify-summary.csv"
+  echo "key,size,bytes,content_type,expected_sha256,actual_sha256,status" > "$VERIFY_SUMMARY"
+  find "$VERIFY_ROWS_DIR" -type f -name '*.csv' -print0 | sort -z | xargs -0 cat >> "$VERIFY_SUMMARY"
+
+  local failed
+  if [[ "$DRY_RUN" == "true" ]]; then
+    echo "Verification dry run complete. Summary: $VERIFY_SUMMARY"
+    return
+  fi
+
+  failed="$(awk -F ',' 'NR > 1 && $7 != "ok" {count++} END {print count + 0}' "$VERIFY_SUMMARY")"
+  local expected actual
+  expected="$(wc -l < "$VERIFY_INPUT" | tr -d ' ')"
+  actual="$(( $(wc -l < "$VERIFY_SUMMARY" | tr -d ' ') - 1 ))"
+  if [[ "$actual" -ne "$expected" ]]; then
+    echo "Verification row count mismatch: expected $expected, got $actual" >&2
+    return 1
+  fi
+  if [[ "$failed" -ne 0 ]]; then
+    echo "Verification failed: $failed object(s). See $VERIFY_SUMMARY" >&2
+    return 1
+  fi
+  echo "Verification passed. Summary: $VERIFY_SUMMARY"
+}
+
+export_functions() {
+  export ENDPOINT ACCESS_KEY SECRET_KEY BUCKET REGION ENCRYPTION SSE_KMS_KEY_ID SSE_C_KEY_FILE AWS_BIN
+  export CLIENT MC_BIN MC_CONFIG_DIR_LOCAL MC_ALIAS
+  export WORK_DIR OUT_DIR WRITE_ROWS_DIR VERIFY_ROWS_DIR DATA_PATTERN KEEP_PAYLOADS DRY_RUN
+  export -f aws_base mc_base aws_cp_args aws_get_args mc_enc_target mc_cp_args mc_get_args payload_path_for_key generate_payload sha256_file sha256_object write_one verify_one
+}
+
+main() {
+  parse_args "$@"
+  validate_args
+  setup_paths
+
+  if [[ "$DRY_RUN" != "true" ]]; then
+    if [[ "$CLIENT" == "mc" ]]; then
+      MC_BIN="$(resolve_mc_bin)"
+      [[ -n "$MC_BIN" ]] || die "mc binary not found; pass --mc-bin or put mc in PATH"
+      require_cmd "$MC_BIN"
+    else
+      require_cmd "$AWS_BIN"
+    fi
+    require_cmd shasum
+    require_cmd head
+    require_cmd yes
+    require_cmd openssl
+  elif [[ "$CLIENT" == "mc" ]]; then
+    MC_BIN="$(resolve_mc_bin)"
+  fi
+
+  setup_mc_alias
+
+  if [[ "$MODE" == "write" || "$MODE" == "mixed" ]]; then
+    local write_failed=false
+    generate_tasks
+    prepare_write_input
+    print_plan
+    create_bucket_if_needed
+    export_functions
+    if ! run_parallel_tasks write "$WRITE_INPUT"; then
+      write_failed=true
+    fi
+    combine_write_manifest
+    echo "Write manifest: $MANIFEST"
+    if [[ "$write_failed" == "true" ]]; then
+      die "one or more write tasks failed; see $OUT_DIR/logs"
+    fi
+  fi
+
+  if [[ "$MODE" == "verify" || "$MODE" == "mixed" ]]; then
+    local verify_failed=false
+    [[ -f "$MANIFEST" ]] || die "manifest not found: $MANIFEST"
+    if [[ "$MODE" == "verify" ]]; then
+      cp "$MANIFEST" "$OUT_DIR/manifest.csv"
+      MANIFEST="$OUT_DIR/manifest.csv"
+    fi
+    prepare_verify_input
+    print_plan
+    export_functions
+    if ! run_parallel_tasks verify "$VERIFY_INPUT"; then
+      verify_failed=true
+    fi
+    combine_verify_summary
+    if [[ "$verify_failed" == "true" ]]; then
+      die "one or more verify tasks failed; see $VERIFY_SUMMARY"
+    fi
+  fi
+}
+
+main "$@"
diff --git a/.dockerignore b/.dockerignore
index eb5a316cb..16224235b 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1 +1,4 @@
 target
+.docker/compat/data
+.docker/compat/kms
+target/compat
diff --git a/.gitignore b/.gitignore
index 21fb4f7cc..0e8bd841b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,6 +52,9 @@ docs/*
 !docs/architecture/
 !docs/architecture/**
 docs/heal-scanner-logging-governance.md
+.codegraph/*
+.docker/compat/data/*
+.docker/compat/kms/*
 
 # nix stuff
 result*
diff --git a/Dockerfile.source b/Dockerfile.source
index f86e6f95e..f9367eacd 100644
--- a/Dockerfile.source
+++ b/Dockerfile.source
@@ -27,6 +27,7 @@
 ARG TARGETPLATFORM
 ARG BUILDPLATFORM
 ARG TARGETARCH
+ARG RUSTFS_BUILD_FEATURES=""
 
 # -----------------------------
 # Build stage
@@ -37,6 +38,7 @@ FROM rust:1.95-trixie AS builder
 ARG TARGETPLATFORM
 ARG BUILDPLATFORM
 ARG TARGETARCH
+ARG RUSTFS_BUILD_FEATURES
 
 # Debug: print platforms
 RUN echo "Build info -> BUILDPLATFORM=${BUILDPLATFORM}, TARGETPLATFORM=${TARGETPLATFORM}"
@@ -143,12 +145,20 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     case "${target_platform}" in \
     linux/amd64) \
     echo "Building for x86_64-unknown-linux-gnu"; \
+    if [ -n "${RUSTFS_BUILD_FEATURES}" ]; then \
+    cargo build --release --locked --target x86_64-unknown-linux-gnu --bin rustfs --features "${RUSTFS_BUILD_FEATURES}" -j "$(nproc)"; \
+    else \
     cargo build --release --locked --target x86_64-unknown-linux-gnu --bin rustfs -j "$(nproc)"; \
+    fi; \
     install -m 0755 "${CARGO_TARGET_DIR}/x86_64-unknown-linux-gnu/release/rustfs" /usr/local/bin/rustfs \
     ;; \
     linux/arm64) \
     echo "Building for aarch64-unknown-linux-gnu"; \
+    if [ -n "${RUSTFS_BUILD_FEATURES}" ]; then \
+    cargo build --release --locked --target aarch64-unknown-linux-gnu --bin rustfs --features "${RUSTFS_BUILD_FEATURES}" -j "$(nproc)"; \
+    else \
     cargo build --release --locked --target aarch64-unknown-linux-gnu --bin rustfs -j "$(nproc)"; \
+    fi; \
     install -m 0755 "${CARGO_TARGET_DIR}/aarch64-unknown-linux-gnu/release/rustfs" /usr/local/bin/rustfs \
     ;; \
     *) \
diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs
index 5a4d2d220..598884e0f 100644
--- a/crates/e2e_test/src/replication_extension_test.rs
+++ b/crates/e2e_test/src/replication_extension_test.rs
@@ -1617,6 +1617,46 @@ async fn test_delete_bucket_replication_removes_remote_target() -> Result<(), Bo
     Ok(())
 }
 
+#[tokio::test]
+#[serial]
+async fn test_bucket_replication_replicates_put_object_issue_2539() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env.start_rustfs_server(vec![]).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "issue-2539-src";
+    let target_bucket = "issue-2539-dst";
+    let object_key = "put-object.txt";
+    let body = "bucket replication should copy PutObject payload";
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(object_key)
+        .body(ByteStream::from(body.as_bytes().to_vec()))
+        .send()
+        .await?;
+
+    wait_for_replicated_object(&target_client, target_bucket, object_key, body).await?;
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_single_bucket_replication_fans_out_to_multiple_targets() -> Result<(), Box<dyn Error + Send + Sync>> {
diff --git a/crates/ecstore/src/bucket/utils.rs b/crates/ecstore/src/bucket/utils.rs
index ceeb3fb1b..8583b7a0a 100644
--- a/crates/ecstore/src/bucket/utils.rs
+++ b/crates/ecstore/src/bucket/utils.rs
@@ -32,7 +32,7 @@ lazy_static::lazy_static! {
 pub fn check_bucket_name_common(bucket_name: &str, strict: bool) -> Result<()> {
     let bucket_name_trimmed = bucket_name.trim();
 
-    if strict && bucket_name_trimmed != bucket_name {
+    if bucket_name_trimmed != bucket_name {
         return Err(Error::other("Bucket name cannot contain leading or trailing whitespace"));
     }
     if bucket_name_trimmed.is_empty() {
@@ -456,10 +456,23 @@ mod tests {
     }
 
     #[test]
-    fn test_strict_bucket_name_rejects_surrounding_whitespace() {
-        assert!(check_valid_bucket_name_strict(" valid-bucket").is_err());
-        assert!(check_valid_bucket_name_strict("valid-bucket ").is_err());
-        assert!(check_valid_bucket_name_strict("\u{c}valid-bucket\u{c}").is_err());
+    fn test_check_bucket_name_rejects_leading_and_trailing_whitespace() {
+        for bucket in [
+            " valid-bucket",
+            "valid-bucket ",
+            "valid-bucket\n",
+            "valid-bucket\u{b}",
+            "\u{c}valid-bucket\u{c}",
+        ] {
+            assert!(
+                check_valid_bucket_name_strict(bucket).is_err(),
+                "bucket name with leading or trailing whitespace must be rejected: {bucket:?}"
+            );
+            assert!(
+                check_valid_bucket_name(bucket).is_err(),
+                "legacy bucket validation must reject leading or trailing whitespace: {bucket:?}"
+            );
+        }
     }
 
     #[test]
diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs
index bc92ea707..f94e695ff 100644
--- a/crates/ecstore/src/object_api/readers.rs
+++ b/crates/ecstore/src/object_api/readers.rs
@@ -97,6 +97,14 @@ fn build_object_encryption_context(
     object_context
 }
 
+#[cfg(feature = "rio-v2")]
+fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool {
+    metadata_get(metadata, INTERNAL_ENCRYPTION_KEY_HEADER).is_some()
+        && metadata_get(metadata, INTERNAL_ENCRYPTION_IV_HEADER).is_some()
+        && metadata_get(metadata, MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER).is_none()
+        && metadata_get(metadata, MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER).is_none()
+}
+
 fn part_plaintext_size(part: &ObjectPartInfo) -> i64 {
     if part.actual_size > 0 {
         part.actual_size
@@ -1301,9 +1309,16 @@ async fn resolve_managed_material(bucket: &str, object: &str, metadata: &HashMap
     let object_context = build_object_encryption_context(bucket, object, kms_context.as_ref());
 
     let decrypted_key = if let Some(service) = get_global_encryption_service().await {
-        service
-            .decrypt_data_key(&encrypted_dek, &object_context)
-            .await
+        #[cfg(feature = "rio-v2")]
+        let data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
+            service.decrypt_legacy_data_key(&encrypted_dek).await
+        } else {
+            service.decrypt_data_key(&encrypted_dek, &object_context).await
+        };
+        #[cfg(not(feature = "rio-v2"))]
+        let data_key = service.decrypt_data_key(&encrypted_dek, &object_context).await;
+
+        data_key
             .map_err(|e| Error::other(format!("failed to decrypt managed data key: {e}")))?
             .plaintext_key
     } else {
@@ -1601,6 +1616,24 @@ mod tests {
         headers
     }
 
+    #[cfg(feature = "rio-v2")]
+    #[test]
+    fn test_legacy_managed_metadata_excludes_sealed_keys() {
+        let legacy_metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "encrypted-dek".to_string()),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "nonce".to_string()),
+        ]);
+        assert!(is_legacy_rustfs_managed_metadata(&legacy_metadata));
+
+        let sealed_metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "encrypted-dek".to_string()),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "nonce".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER.to_string(), "sealed-key".to_string()),
+        ]);
+
+        assert!(!is_legacy_rustfs_managed_metadata(&sealed_metadata));
+    }
+
     #[cfg(feature = "rio-v2")]
     fn seal_ssec_object_key_for_test(
         bucket: &str,
diff --git a/crates/kms/src/service.rs b/crates/kms/src/service.rs
index 4e7501328..a7f91c645 100644
--- a/crates/kms/src/service.rs
+++ b/crates/kms/src/service.rs
@@ -232,9 +232,26 @@ impl ObjectEncryptionService {
     /// DataKey with decrypted key
     ///
     pub async fn decrypt_data_key(&self, encrypted_key: &[u8], context: &ObjectEncryptionContext) -> Result<DataKey> {
+        self.decrypt_data_key_with_context(encrypted_key, request_encryption_context(context))
+            .await
+    }
+
+    /// Decrypt a data key written by legacy RustFS versions that reused KMS data
+    /// keys across objects with different encryption contexts.
+    ///
+    /// Callers must restrict this to positively identified legacy object metadata.
+    pub async fn decrypt_legacy_data_key(&self, encrypted_key: &[u8]) -> Result<DataKey> {
+        self.decrypt_data_key_with_context(encrypted_key, HashMap::new()).await
+    }
+
+    async fn decrypt_data_key_with_context(
+        &self,
+        encrypted_key: &[u8],
+        encryption_context: HashMap<String, String>,
+    ) -> Result<DataKey> {
         let decrypt_request = DecryptRequest {
             ciphertext: encrypted_key.to_vec(),
-            encryption_context: request_encryption_context(context),
+            encryption_context,
             grant_tokens: Vec::new(),
         };
 
@@ -928,5 +945,11 @@ mod tests {
             .await
             .expect("decrypt should accept matching KMS context");
         assert_ne!(decrypted.plaintext_key, [0u8; 32]);
+
+        let legacy_decrypted = service
+            .decrypt_legacy_data_key(&encrypted_key)
+            .await
+            .expect("legacy decrypt should use the backend compatibility path");
+        assert_eq!(legacy_decrypted.plaintext_key, decrypted.plaintext_key);
     }
 }
diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs
index c7f786dce..5d17715f7 100644
--- a/rustfs/src/storage/sse.rs
+++ b/rustfs/src/storage/sse.rs
@@ -1680,10 +1680,22 @@ async fn apply_managed_decryption_material(
 
     // Use factory pattern to get provider (test or production mode)
     let provider = get_sse_dek_provider().await?;
+    #[cfg(feature = "rio-v2")]
+    let decrypted_data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
+        provider
+            .decrypt_legacy_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
+            .await
+    } else {
+        provider
+            .decrypt_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
+            .await
+    };
+    #[cfg(not(feature = "rio-v2"))]
     let decrypted_data_key = provider
         .decrypt_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
-        .await
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {e}"))))?;
+        .await;
+    let decrypted_data_key =
+        decrypted_data_key.map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {e}"))))?;
     #[cfg(feature = "rio-v2")]
     let (key_bytes, base_nonce, key_kind) = if let Some(sealed_key) = minio_sealed_key {
         (
@@ -1761,6 +1773,17 @@ pub trait SseDekProvider: Send + Sync {
         kms_key_id: &str,
         context: &ObjectEncryptionContext,
     ) -> Result<[u8; 32], ApiError>;
+
+    /// Decrypt a DEK from positively identified legacy managed metadata.
+    #[cfg(feature = "rio-v2")]
+    async fn decrypt_legacy_sse_dek(
+        &self,
+        encrypted_dek: &[u8],
+        kms_key_id: &str,
+        context: &ObjectEncryptionContext,
+    ) -> Result<[u8; 32], ApiError> {
+        self.decrypt_sse_dek(encrypted_dek, kms_key_id, context).await
+    }
 }
 
 // ============================================================================
@@ -1820,6 +1843,24 @@ impl SseDekProvider for KmsSseDekProvider {
 
         Ok(data_key.plaintext_key)
     }
+
+    #[cfg(feature = "rio-v2")]
+    async fn decrypt_legacy_sse_dek(
+        &self,
+        encrypted_dek: &[u8],
+        _kms_key_id: &str,
+        _context: &ObjectEncryptionContext,
+    ) -> Result<[u8; 32], ApiError> {
+        let service = Self::current_service()
+            .await
+            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
+        let data_key = service
+            .decrypt_legacy_data_key(encrypted_dek)
+            .await
+            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt legacy data key: {e}"))))?;
+
+        Ok(data_key.plaintext_key)
+    }
 }
 
 // ============================================================================
@@ -2139,6 +2180,14 @@ fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> b
         || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
 }
 
+#[cfg(feature = "rio-v2")]
+fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool {
+    metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
+        && metadata.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)
+        && !metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER)
+        && !metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
+}
+
 #[cfg(feature = "rio-v2")]
 fn parse_minio_managed_sealed_key(
     metadata: &HashMap<String, String>,
@@ -3047,6 +3096,23 @@ mod tests {
         assert!(metadata.contains_key("content-type"));
     }
 
+    #[cfg(feature = "rio-v2")]
+    #[test]
+    fn test_legacy_managed_metadata_excludes_sealed_keys() {
+        let legacy_metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "encrypted-dek".to_string()),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "nonce".to_string()),
+        ]);
+        assert!(is_legacy_rustfs_managed_metadata(&legacy_metadata));
+
+        let sealed_metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "encrypted-dek".to_string()),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "nonce".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER.to_string(), "sealed-key".to_string()),
+        ]);
+        assert!(!is_legacy_rustfs_managed_metadata(&sealed_metadata));
+    }
+
     #[cfg(feature = "rio-v2")]
     #[test]
     fn test_normalize_managed_metadata_accepts_minio_only_headers() {