fix(startup): avoid panic when system CA roots are missing

fix(api): preserve server-side storage error surface (#6753 )
Co-authored-by: heihutu <[email protected]>
2026-08-28 05:50:22 +08:00 · 2026-08-27 15:42:51 +00:00 · 2026-08-27 22:55:26 +08:00 · 2026-08-27 22:34:10 +08:00 · 2026-08-27 22:25:33 +08:00 · 2026-08-27 22:25:17 +08:00
17 changed files with 2438 additions and 103 deletions
@@ -15,10 +15,6 @@ on:
        description: 'Stop warp when surviving nodes reach N GiB'
        required: false
        default: '40'
-      heal_target_gb:
-        description: 'Outage node must reach N GiB after heal to pass'
-        required: false
-        default: '40'
      cleanup_before:
        description: 'Reset the nodes before the test (DESTROYS existing data/config)'
        type: boolean
@@ -100,7 +96,6 @@ jobs:
            --endpoint "${{ env.RUSTFS_API_ENDPOINT }}" \
            --stop-node-gb "${{ inputs.stop_node_gb }}" \
            --warp-stop-gb "${{ inputs.warp_stop_gb }}" \
-            --heal-target-gb "${{ inputs.heal_target_gb }}" \
            --log-file /tmp/rustfs-heal-test.log

      - name: Upload test logs
@@ -0,0 +1,203 @@
+name: RustFS Performance Test
+
+on:
+  workflow_dispatch:
+    inputs:
+      package_url:
+        description: 'Direct .deb URL (nightly/R2). Defaults to the latest nightly deb.'
+        required: false
+        type: string
+      test_method:
+        description: 'Benchmark method(s) to run (manual runs only; "all" = GET+PUT+MIXED)'
+        type: choice
+        options:
+          - all
+          - get
+          - put
+          - mixed
+        default: 'all'
+      object_size:
+        description: 'Object size(s) to test (manual runs only; "all" = all 10 sizes)'
+        type: choice
+        options:
+          - all
+          - 1KiB
+          - 4KiB
+          - 16KiB
+          - 128KiB
+          - 1MiB
+          - 4MiB
+          - 8MiB
+          - 16MiB
+          - 32MiB
+          - 64MiB
+        default: 'all'
+      warp_duration:
+        description: 'warp duration per round (e.g. 5m, 30s)'
+        required: false
+        default: '5m'
+      warp_concurrency:
+        description: 'warp concurrency'
+        required: false
+        default: '64'
+      cleanup_before:
+        description: 'Reset the nodes before the test (DESTROYS existing data/config)'
+        type: boolean
+        default: true
+      cleanup_after:
+        description: 'Reset the nodes after the test (DESTROYS test data/config)'
+        type: boolean
+        default: true
+  workflow_run:
+    # Run after the nightly build completes; the nightly deb is what the test installs.
+    workflows: ["Nightly GNU Build"]
+    types: [completed]
+
+permissions:
+  contents: read
+
+# Dedicated pf-testing runner/environment: own concurrency group so perf runs
+# never block (or are blocked by) the pool-expansion / heal tests.
+concurrency:
+  group: rustfs-performance-test
+  cancel-in-progress: false
+
+defaults:
+  run:
+    shell: bash
+
+env:
+  RUSTFS_ACCESS_KEY: ${{ secrets.RUSTFS_ACCESS_KEY }}
+  RUSTFS_SECRET_KEY: ${{ secrets.RUSTFS_SECRET_KEY }}
+  # Performance test uses its own node list (4 nodes); the shared
+  # RUSTFS_NODES secret is used by the 3-node pool-expansion / heal tests.
+  RUSTFS_NODES: ${{ secrets.RUSTFS_PERF_NODES || vars.RUSTFS_PERF_NODES || 'vm000 vm001 vm002 vm003' }}
+  RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER }}
+  # Package used by the nightly run (workflow_dispatch inputs are empty for
+  # workflow_run events), i.e. the latest nightly deb published by nightly-gnu.yml.
+  RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
+  # Fixed benchmark result directory so later steps can read summary.md
+  RUSTFS_RESULT_DIR: /tmp/rustfs-perf-results
+  # Cross-repo token for writing to rustfs/backlog (set in repo settings)
+  PF_TESTING_GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
+
+jobs:
+  performance-test:
+    runs-on: pf-testing
+    timeout-minutes: 900
+    # Run on manual dispatch, or when the nightly build completed successfully.
+    # Skipped when nightly failed.
+    if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+          ref: ${{ github.event.workflow_run.head_sha || github.ref }}
+
+      - name: Show environment
+        run: |
+          uname -a
+          jq --version
+          warp --version || true
+          df -h /data | tail -1
+
+      - name: Reset test environment (before)
+        if: ${{ inputs.cleanup_before != 'false' }}
+        run: |
+          chmod +x scripts/test/rustfs_performance_test.sh
+          ./scripts/test/rustfs_performance_test.sh --step 1 -y
+
+      - name: Install RustFS package & start cluster (4x4)
+        run: |
+          ARGS=(--steps "2,3,4" -y)
+          if [ -n "${{ inputs.package_url }}" ]; then
+            ARGS+=(--package-url "${{ inputs.package_url }}")
+          else
+            ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
+          fi
+          ./scripts/test/rustfs_performance_test.sh "${ARGS[@]}"
+
+      - name: Preflight checks
+        run: |
+          ARGS=(--preflight)
+          if [ -n "${{ inputs.package_url }}" ]; then
+            ARGS+=(--package-url "${{ inputs.package_url }}")
+          else
+            ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
+          fi
+          ./scripts/test/rustfs_performance_test.sh "${ARGS[@]}"
+
+      - name: Run benchmark (GET/PUT/MIXED)
+        id: benchmark
+        run: |
+          # Empty on automatic (workflow_run) runs -> full 30 rounds.
+          # Manual dispatch can restrict method(s)/size(s).
+          export WARP_METHODS="${{ inputs.test_method }}"
+          export WARP_SIZES="${{ inputs.object_size }}"
+          ./scripts/test/rustfs_performance_test.sh \
+            --step 5 -y \
+            --warp-duration "${{ inputs.warp_duration || '5m' }}" \
+            --warp-concurrency "${{ inputs.warp_concurrency || '64' }}" \
+            --log-file /tmp/rustfs-perf-test.log
+
+      - name: Analyze results
+        if: ${{ steps.benchmark.conclusion == 'success' }}
+        run: |
+          ./scripts/test/rustfs_performance_test.sh --step 6 -y
+
+      - name: Post results to backlog issue
+        if: ${{ steps.benchmark.conclusion == 'success' }}
+        env:
+          GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          RESULT_DIR: ${{ env.RUSTFS_RESULT_DIR }}
+        run: |
+          set -euo pipefail
+          if [ -z "${GH_TOKEN:-}" ]; then
+            echo "PF_TESTING_GH_TOKEN is not configured; skipping issue post"
+            exit 0
+          fi
+          SUMMARY="${RESULT_DIR}/summary.md"
+          [ -f "${SUMMARY}" ] || { echo "summary.md not found at ${SUMMARY}"; exit 1; }
+          DATE="$(date -u +%Y-%m-%d)"
+          {
+            echo "## RustFS nightly build performance testing report"
+            echo ""
+            echo "- **日期**: ${DATE}"
+            echo "- **Run**: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
+            echo "- **触发方式**: ${{ github.event_name }}"
+            echo ""
+            cat "${SUMMARY}"
+          } > /tmp/rustfs-perf-issue-body.md
+          TITLE="RustFS nightly build performance testing report"
+          EXISTING="$(gh issue list --repo rustfs/backlog \
+            --search "in:title \"${TITLE}\"" --state all --limit 5 \
+            --json number --jq '.[0].number // empty')"
+          if [ -n "${EXISTING}" ]; then
+            gh issue comment "${EXISTING}" --repo rustfs/backlog --body-file /tmp/rustfs-perf-issue-body.md
+            echo "commented on existing issue #${EXISTING}"
+          else
+            gh issue create --repo rustfs/backlog --title "${TITLE}" --body-file /tmp/rustfs-perf-issue-body.md
+          fi
+
+      - name: Upload test logs & results
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: rustfs-perf-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-perf-test*.log
+            /tmp/rustfs-perf-results/**
+          if-no-files-found: warn
+
+      - name: Reset test environment (after)
+        if: ${{ always() && inputs.cleanup_after != 'false' }}
+        run: |
+          ./scripts/test/rustfs_performance_test.sh --step 7 -y
+
+      - name: Notify on failure
+        if: failure()
+        run: |
+          echo "RustFS performance test failed"
+          echo "Package source: ${{ inputs.package_url || 'nightly (R2 latest)' }}"
+          echo "See the uploaded log artifact for details."
@@ -38,10 +38,6 @@ on:
        description: 'Heal: stop warp when surviving nodes reach N GiB'
        required: false
        default: '40'
-      heal_target_gb:
-        description: 'Heal: outage node must reach N GiB after heal'
-        required: false
-        default: '40'
      cleanup_before:
        description: 'Reset the nodes before the test (DESTROYS existing data/config)'
        type: boolean
@@ -218,7 +214,6 @@ jobs:
            --endpoint "${{ env.RUSTFS_API_ENDPOINT }}" \
            --stop-node-gb "${{ inputs.stop_node_gb || '15' }}" \
            --warp-stop-gb "${{ inputs.warp_stop_gb || '40' }}" \
-            --heal-target-gb "${{ inputs.heal_target_gb || '40' }}" \
            --log-file /tmp/rustfs-heal-test.log

      - name: Upload test logs
@@ -43,7 +43,7 @@ catalog extension.
 | PyIceberg | Automated | Creates namespace and table, appends rows, reloads, scans, probes metadata-location, refs, views, maintenance, diagnostics, and optional catalog-vended table credentials with an exact-prefix data-plane scope check. |
 | Spark Iceberg REST catalog | Manual/live harness | RustFS can generate pinned Spark/Iceberg package inputs, REST catalog properties, SQL, run commands, expected `row_count=2`, and a CI opt-in gate for namespace creation, table creation, append, refresh, count, and cleanup. Live Spark execution and commit-conflict probing are still manual validation items unless explicitly enabled in the runner. |
 | Trino Iceberg REST catalog | Manual/live harness | RustFS can generate catalog properties and a read-only `SELECT COUNT(*)` command for a table created by PyIceberg or Spark. Write compatibility is not claimed. |
-| DuckDB Iceberg | Manual/live harness | RustFS can generate `httpfs` and `iceberg` SQL using an operator-supplied current metadata location. Write and commit compatibility are not claimed. |
+| DuckDB Iceberg 1.5.5 | Automated | `duckdb_smoke.py` verifies the metadata-location read path and generic REST Catalog single-table create, insert, update, delete, merge, schema evolution, snapshots, concurrent writers, normal drop, PyIceberg cross-read, `/iceberg` with `s3` signing, and `/_iceberg` with `s3tables` signing. Staged create, purge-on-drop, and format v3 are verified as fail-closed boundaries. DuckDB's endpoint-disabled two-table mode is exercised without claiming cross-table atomicity. AWS `ENDPOINT_TYPE S3_TABLES` and catalog-vended credential integration are not claimed. |
 | StarRocks Iceberg REST catalog | Documented, not automated | External catalog read-path reference only. Write compatibility is not claimed. |
 | Databend | Manual/live harness | RustFS can generate an S3 stage read probe for table data files. RustFS does not claim Databend Iceberg REST Catalog integration yet. |
 | Snowflake Open Catalog / Iceberg integrations | Generated harness | RustFS can generate an operator-adapted external volume/catalog SQL template. Live RustFS interoperability is not claimed. |
@@ -52,10 +52,10 @@ catalog extension.

 | Area | Status | Current RustFS claim |
 |---|---|---|
-| Live conformance evidence | Manual/live harness | `engine_compatibility.py --print-live-evidence-schema` defines the required evidence schema and claim promotion boundaries. `pyiceberg_smoke.py --live-evidence-output` writes a validated PyIceberg evidence record after a successful live smoke run. |
+| Live conformance evidence | Automated for PyIceberg and DuckDB | `engine_compatibility.py --print-live-evidence-schema` defines the required evidence schema and claim promotion boundaries. `pyiceberg_smoke.py --live-evidence-output` and `duckdb_smoke.py --live-evidence-output` write validated client evidence records after successful live smoke runs. |
 | Production operations guide | Generated harness | `engine_compatibility.py --print-operations-guide` records command, evidence, pass criteria, and fail-closed signals for live conformance, durable backing cutover, maintenance, recovery, permissions, credential vending, and unsupported-claim governance. |
 | Vendor compatibility gap audit | Generated harness | `engine_compatibility.py --print-vendor-audit` records provider source URLs, catalog path and warehouse shapes, signing/auth models, error/permission/maintenance validation categories, and not-claimed boundaries for AWS S3 Tables, MinIO AIStor Tables, Cloudflare R2 Data Catalog, and Alibaba OSS Tables. |
-| Client claim promotion | Documented, not automated | PyIceberg remains the automated claim. Spark can be promoted only with recorded manual/live evidence; Trino and DuckDB read probes do not promote write compatibility; Snowflake and vendor profiles remain reference-only without repeatable live evidence. |
+| Client claim promotion | Automated for scoped clients | PyIceberg and DuckDB claims remain bounded by their repeatable smoke entrypoints and recorded versions. Spark can be promoted only with recorded manual/live evidence; Trino remains read-only; Snowflake and vendor profiles remain reference-only without repeatable live evidence. |

 ## Catalog API Matrix

@@ -242,6 +242,7 @@ compatibility claims:
 ```bash
 python3 scripts/table-catalog/test_pyiceberg_smoke.py
 python3 scripts/table-catalog/test_engine_compatibility.py
+python3 scripts/table-catalog/test_duckdb_smoke.py
 python3 scripts/table-catalog/test_failure_coverage.py
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-client-matrix
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-engine-compatibility
@@ -250,6 +251,7 @@ python3 scripts/table-catalog/pyiceberg_smoke.py --print-vendor-profiles
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-readiness
 python3 scripts/table-catalog/engine_compatibility.py --print-vendor-audit
 python3 scripts/table-catalog/engine_compatibility.py --print-spark-config
+python3 scripts/table-catalog/engine_compatibility.py --print-duckdb-rest-sql
 python3 scripts/table-catalog/engine_compatibility.py \
  --profile aws-s3tables \
  --region us-east-1 \
@@ -304,9 +306,9 @@ Use conservative release wording that matches the matrix.
 Acceptable wording:

 > RustFS includes a core Iceberg REST Catalog-based S3 Tables implementation
-> with PyIceberg smoke coverage, table-aware S3 data-plane policy checks,
+> with PyIceberg and DuckDB smoke coverage, table-aware S3 data-plane policy checks,
 > controlled maintenance, catalog recovery diagnostics, manual conformance
-> input for Spark, Trino, DuckDB, Databend, and Snowflake, production-failure
+> input for Spark, Trino, Databend, and Snowflake, production-failure
 > probe harnesses, disaster-recovery and scale/fault rehearsal probes, and a
 > machine-readable production operations evidence guide.

@@ -17,6 +17,71 @@
 use super::*;

 use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation};
+use crate::error::ServerSideSourceReadError;
+
+struct CopySourceReadStream<R> {
+    inner: R,
+    remaining: i64,
+}
+
+impl<R> CopySourceReadStream<R> {
+    fn new(inner: R, expected_size: i64) -> Self {
+        Self {
+            inner,
+            remaining: expected_size.max(0),
+        }
+    }
+}
+
+fn copy_source_read_stream<R>(inner: R, expected_size: i64) -> CopySourceReadStream<R> {
+    CopySourceReadStream::new(inner, expected_size)
+}
+
+fn copy_source_read_error(source: std::io::Error) -> std::io::Error {
+    let kind = source.kind();
+    std::io::Error::new(kind, ServerSideSourceReadError::new("CopyObject", source))
+}
+
+fn copy_source_incomplete_body_error(remaining: i64) -> std::io::Error {
+    copy_source_read_error(std::io::Error::new(
+        std::io::ErrorKind::UnexpectedEof,
+        rustfs_rio::IncompleteBody { remaining },
+    ))
+}
+
+impl<R> AsyncRead for CopySourceReadStream<R>
+where
+    R: AsyncRead + Unpin,
+{
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        let before = buf.filled().len();
+        match Pin::new(&mut this.inner).poll_read(cx, buf) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Err(err)) => Poll::Ready(Err(copy_source_read_error(err))),
+            Poll::Ready(Ok(())) => {
+                let read = buf.filled().len() - before;
+                if read == 0 {
+                    if this.remaining > 0 {
+                        return Poll::Ready(Err(copy_source_incomplete_body_error(this.remaining)));
+                    }
+                    return Poll::Ready(Ok(()));
+                }
+
+                let read = match i64::try_from(read) {
+                    Ok(read) => read,
+                    Err(_) => {
+                        return Poll::Ready(Err(copy_source_read_error(std::io::Error::other(
+                            "copy source read count exceeds i64::MAX",
+                        ))));
+                    }
+                };
+                this.remaining = this.remaining.saturating_sub(read);
+                Poll::Ready(Ok(()))
+            }
+        }
+    }
+}

 fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError {
    match err {
@@ -580,11 +645,13 @@ impl DefaultObjectUsecase {
        let mut write_plan = WritePlan::new();
        let mut reader = if should_compress {
            let algorithm = CompressionAlgorithm::default();
-            let hrd = HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)?;
+            let hrd = HashReader::from_stream(copy_source_read_stream(gr.stream, length), length, actual_size, None, None, false)
+                .map_err(ApiError::from)?;
            write_plan = write_plan.with_compression(algorithm);
            hrd
        } else {
-            HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)?
+            HashReader::from_stream(copy_source_read_stream(gr.stream, length), length, actual_size, None, None, false)
+                .map_err(ApiError::from)?
        };

        // Give the destination object a checksum so CopyObject returns it and a later checksum-mode
@@ -835,6 +902,7 @@ mod tests {
    use http::{HeaderValue, Method};
    use s3s::dto::{ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule};
    use std::sync::Arc;
+    use tokio::io::AsyncReadExt;

    // A malformed bucket-default algorithm reaches this resolution only through
    // corrupt or hand-edited bucket metadata (PutBucketEncryption validates the
@@ -876,6 +944,23 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn copy_source_read_stream_maps_short_eof_to_service_unavailable() {
+        let source = std::io::Cursor::new(b"abc".to_vec());
+        let mut reader = HashReader::from_stream(copy_source_read_stream(source, 4), 4, 4, None, None, false)
+            .expect("copy source hash reader should build");
+
+        let mut output = Vec::new();
+        let err = reader
+            .read_to_end(&mut output)
+            .await
+            .expect_err("short copy source must fail before destination write succeeds");
+        let api_error = ApiError::from(err);
+
+        assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable);
+        assert_ne!(api_error.code, S3ErrorCode::IncompleteBody);
+    }
+
    #[tokio::test]
    async fn execute_copy_object_rejects_self_copy_without_replace_directive() {
        let input = CopyObjectInput::builder()
@@ -34,6 +34,32 @@ impl std::fmt::Display for UploadLimitExceeded {

 impl std::error::Error for UploadLimitExceeded {}

+/// Marks a server-side object/source reader failure that must not be reported as
+/// a malformed client request body.
+#[derive(Debug)]
+pub(crate) struct ServerSideSourceReadError {
+    operation: &'static str,
+    source: std::io::Error,
+}
+
+impl ServerSideSourceReadError {
+    pub(crate) const fn new(operation: &'static str, source: std::io::Error) -> Self {
+        Self { operation, source }
+    }
+}
+
+impl std::fmt::Display for ServerSideSourceReadError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} source read failed: {}", self.operation, self.source)
+    }
+}
+
+impl std::error::Error for ServerSideSourceReadError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(&self.source)
+    }
+}
+
 #[derive(Debug)]
 pub struct ApiError {
    pub code: S3ErrorCode,
@@ -302,6 +328,17 @@ impl From<StorageError> for ApiError {
            };
        }

+        if let StorageError::Io(ref io_err) = err
+            && let Some(inner) = io_err.get_ref()
+            && error_chain_has_type::<ServerSideSourceReadError>(inner)
+        {
+            return ApiError {
+                code: S3ErrorCode::ServiceUnavailable,
+                message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
+                source: Some(Box::new(err)),
+            };
+        }
+
        if let StorageError::Io(ref io_err) = err
            && io_err
                .get_ref()
@@ -340,15 +377,15 @@ impl From<StorageError> for ApiError {
            StorageError::ObjectNameInvalid(_, _) => S3ErrorCode::InvalidArgument,
            StorageError::BucketExists(_) => S3ErrorCode::BucketAlreadyOwnedByYou,
            StorageError::StorageFull => S3ErrorCode::ServiceUnavailable,
-            StorageError::SlowDown
-            | StorageError::FaultyDisk
+            StorageError::SlowDown => S3ErrorCode::SlowDown,
+            StorageError::FaultyDisk
            | StorageError::FaultyRemoteDisk
            | StorageError::DiskNotFound
-            | StorageError::TooManyOpenFiles => S3ErrorCode::SlowDown,
+            | StorageError::TooManyOpenFiles => S3ErrorCode::ServiceUnavailable,
            StorageError::ErasureReadQuorum
            | StorageError::InsufficientReadQuorum(_, _)
            | StorageError::ErasureWriteQuorum
-            | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::SlowDown,
+            | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::ServiceUnavailable,
            StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable,
            StorageError::QuotaExceeded { .. } => S3ErrorCode::InvalidRequest,
            StorageError::Lock(_) => S3ErrorCode::ServiceUnavailable,
@@ -435,6 +472,13 @@ impl From<std::io::Error> for ApiError {
                    source: Some(Box::new(err)),
                };
            }
+            if error_chain_has_type::<ServerSideSourceReadError>(inner) {
+                return ApiError {
+                    code: S3ErrorCode::ServiceUnavailable,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
+                    source: Some(Box::new(err)),
+                };
+            }
            if error_chain_has_type::<rustfs_rio::IncompleteBody>(inner) {
                return ApiError {
                    code: S3ErrorCode::IncompleteBody,
@@ -613,6 +657,22 @@ mod tests {
        }
    }

+    #[test]
+    fn server_side_source_read_error_maps_to_service_unavailable_before_incomplete_body() {
+        let short_source = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 17 });
+        let marker = ServerSideSourceReadError::new("CopyObject", short_source);
+
+        let api_error = ApiError::from(IoError::new(ErrorKind::UnexpectedEof, marker));
+        assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable);
+        assert_ne!(api_error.code, S3ErrorCode::IncompleteBody);
+
+        let short_source = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 17 });
+        let marker = ServerSideSourceReadError::new("CopyObject", short_source);
+        let api_error = ApiError::from(StorageError::Io(IoError::new(ErrorKind::UnexpectedEof, marker)));
+        assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable);
+        assert_ne!(api_error.code, S3ErrorCode::IncompleteBody);
+    }
+
    #[test]
    fn test_api_error_surfaces_invalid_argument_reason() {
        let err = StorageError::InvalidArgument(
@@ -785,14 +845,20 @@ mod tests {
            (StorageError::BucketExists("test".into()), S3ErrorCode::BucketAlreadyOwnedByYou),
            (StorageError::StorageFull, S3ErrorCode::ServiceUnavailable),
            (StorageError::SlowDown, S3ErrorCode::SlowDown),
-            (StorageError::FaultyDisk, S3ErrorCode::SlowDown),
-            (StorageError::FaultyRemoteDisk, S3ErrorCode::SlowDown),
-            (StorageError::DiskNotFound, S3ErrorCode::SlowDown),
-            (StorageError::TooManyOpenFiles, S3ErrorCode::SlowDown),
-            (StorageError::ErasureReadQuorum, S3ErrorCode::SlowDown),
-            (StorageError::InsufficientReadQuorum("test".into(), "test".into()), S3ErrorCode::SlowDown),
-            (StorageError::ErasureWriteQuorum, S3ErrorCode::SlowDown),
-            (StorageError::InsufficientWriteQuorum("test".into(), "test".into()), S3ErrorCode::SlowDown),
+            (StorageError::FaultyDisk, S3ErrorCode::ServiceUnavailable),
+            (StorageError::FaultyRemoteDisk, S3ErrorCode::ServiceUnavailable),
+            (StorageError::DiskNotFound, S3ErrorCode::ServiceUnavailable),
+            (StorageError::TooManyOpenFiles, S3ErrorCode::ServiceUnavailable),
+            (StorageError::ErasureReadQuorum, S3ErrorCode::ServiceUnavailable),
+            (
+                StorageError::InsufficientReadQuorum("test".into(), "test".into()),
+                S3ErrorCode::ServiceUnavailable,
+            ),
+            (StorageError::ErasureWriteQuorum, S3ErrorCode::ServiceUnavailable),
+            (
+                StorageError::InsufficientWriteQuorum("test".into(), "test".into()),
+                S3ErrorCode::ServiceUnavailable,
+            ),
            (
                StorageError::NamespaceLockQuorumUnavailable {
                    mode: "write",
@@ -59,8 +59,6 @@ pub struct UpdateCheckResult {

 /// Version checker
 pub struct VersionChecker {
-    /// HTTP client
-    client: reqwest::Client,
    /// Version server URL
    version_url: String,
    /// Request timeout
@@ -76,14 +74,7 @@ impl Default for VersionChecker {
 impl VersionChecker {
    /// Create a new version checker
    pub fn new() -> Self {
-        let client = reqwest::Client::builder()
-            .timeout(Duration::from_secs(10))
-            .user_agent(format!("RustFS/{}", get_current_version()))
-            .build()
-            .unwrap_or_else(|_| reqwest::Client::new());
-
        Self {
-            client,
            version_url: "https://version.rustfs.com/latest.json".to_string(),
            timeout: Duration::from_secs(10),
        }
@@ -91,14 +82,7 @@ impl VersionChecker {

    /// Create version checker with custom configuration
    pub fn with_config(url: String, timeout: Duration) -> Self {
-        let client = reqwest::Client::builder()
-            .timeout(timeout)
-            .user_agent(format!("RustFS/{}", get_current_version()))
-            .build()
-            .unwrap_or_else(|_| reqwest::Client::new());
-
        Self {
-            client,
            version_url: url,
            timeout,
        }
@@ -108,9 +92,13 @@ impl VersionChecker {
    pub async fn check_for_updates(&self) -> Result<UpdateCheckResult, UpdateCheckError> {
        let current_version = get_current_version();
        debug!("Checking for updates, current version: {}", current_version);
+        let client = reqwest::Client::builder()
+            .timeout(self.timeout)
+            .user_agent(format!("RustFS/{current_version}"))
+            .build()?;

        // Send HTTP GET request to get latest version information
-        let response = self.client.get(&self.version_url).timeout(self.timeout).send().await?;
+        let response = client.get(&self.version_url).timeout(self.timeout).send().await?;

        if !response.status().is_success() {
            let status = response.status();
@@ -182,6 +170,32 @@ pub async fn check_updates_with_url(url: String) -> Result<UpdateCheckResult, Up
 mod tests {
    use super::*;

+    #[tokio::test]
+    #[serial_test::serial]
+    async fn version_checker_construction_does_not_require_system_roots() {
+        #[cfg(target_os = "linux")]
+        {
+            let temp = tempfile::tempdir().expect("temporary certificate directory");
+            let cert_file = temp.path().join("empty.pem");
+            std::fs::write(&cert_file, []).expect("empty certificate file");
+            let cert_file = cert_file.to_string_lossy().into_owned();
+            let result =
+                temp_env::async_with_vars([("SSL_CERT_FILE", Some(cert_file.as_str())), ("SSL_CERT_DIR", Some(""))], async {
+                    let checker = VersionChecker::new();
+                    checker.check_for_updates().await
+                })
+                .await;
+            assert!(matches!(result, Err(UpdateCheckError::HttpError(_))));
+        }
+
+        #[cfg(not(target_os = "linux"))]
+        {
+            let checker = VersionChecker::new();
+            assert_eq!(checker.version_url, "https://version.rustfs.com/latest.json");
+            assert_eq!(checker.timeout, Duration::from_secs(10));
+        }
+    }
+
    #[tokio::test]
    async fn test_get_current_version() {
        let version = get_current_version();
@@ -163,10 +163,10 @@ python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-sche
 ```

 Use these outputs when updating release notes, PR descriptions, or follow-up
-work items. They are intentionally conservative: only PyIceberg is automated by
-this script today. Spark has a repeatable manual/live harness with pinned
-client package inputs, generated configuration, generated SQL, expected
-results, and a CI opt-in gate. Trino, DuckDB, Databend, and Snowflake now have
+work items. They are intentionally conservative: PyIceberg and DuckDB have
+separate automated smoke entrypoints. Spark has a repeatable manual/live harness
+with pinned client package inputs, generated configuration, generated SQL,
+expected results, and a CI opt-in gate. Trino, Databend, and Snowflake have
 generated manual probe inputs, but they remain opt-in and do not promote write
 or full vendor interoperability claims.

@@ -241,6 +241,7 @@ python3 scripts/table-catalog/engine_compatibility.py \
  --table-bucket analytics \
  --print-spark-config
 python3 scripts/table-catalog/engine_compatibility.py --print-spark-sql --cleanup
+python3 scripts/table-catalog/engine_compatibility.py --print-duckdb-rest-sql
 python3 scripts/table-catalog/engine_compatibility.py --print-live-conformance --cleanup
 python3 scripts/table-catalog/engine_compatibility.py --print-operations-guide
 ```
@@ -310,7 +311,7 @@ The smoke test also probes catalog-backed advanced Iceberg surfaces:
 | PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional catalog-vended table credentials with exact-prefix data-plane scope probe |
 | Spark Iceberg REST catalog | Manual/live harness | pinned Spark and Iceberg package inputs, configuration, SQL, run command, expected row count, and cleanup can be generated for a running RustFS endpoint; CI execution is opt-in |
 | Trino Iceberg REST catalog | Manual/live read probe | generated catalog properties and a read-only SELECT probe for a table created by PyIceberg or Spark; no write compatibility claim yet |
-| DuckDB Iceberg | Manual/live read probe | generated httpfs/iceberg SQL using an operator-supplied current metadata location; read-path only |
+| DuckDB Iceberg | Automated smoke target | metadata-location read plus generic REST Catalog single-table DDL, DML, schema evolution, snapshots, `/iceberg` and `/_iceberg` signing, fail-closed unsupported boundaries, endpoint-disabled non-atomic multi-table mode, concurrent writers, and PyIceberg cross-read |
 | StarRocks Iceberg REST catalog | Documented, not automated | external catalog read-path reference only |
 | Databend | Manual/live S3 stage probe | generated S3 stage read probe for table data files; Iceberg REST catalog integration is not claimed |
 | Snowflake/Open Catalog integrations | Manual reference probe | generated external volume/catalog SQL template; live RustFS interoperability is not claimed |
@@ -503,6 +504,70 @@ RUSTFS_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS=900

 The TTL is clamped to the supported short-lived range by the server.

+## DuckDB REST Catalog Profile
+
+DuckDB can read an individual Iceberg table with `iceberg_scan` or attach RustFS
+as a generic Iceberg REST Catalog. The metadata-location path remains read-only.
+The attached catalog path is the prerequisite for DuckDB writes.
+
+Generate the canonical RustFS REST Catalog profile:
+
+```bash
+python3 scripts/table-catalog/engine_compatibility.py \
+  --endpoint http://127.0.0.1:9000 \
+  --warehouse rustfs-s3table-smoke \
+  --namespace smoke \
+  --table events \
+  --rest-path /iceberg \
+  --rest-signing-name s3 \
+  --print-duckdb-rest-sql
+```
+
+Generate the compatibility alias profile by changing the last three arguments:
+
+```bash
+python3 scripts/table-catalog/engine_compatibility.py \
+  --rest-path /_iceberg \
+  --rest-signing-name s3tables \
+  --print-duckdb-rest-sql
+```
+
+The generated `ATTACH` disables staged create, post-create metadata updates,
+multi-table commit, client-side file removal, and purge-on-drop. These options
+keep DuckDB within RustFS's claimed single-table REST surface. Do not replace
+the explicit endpoint with DuckDB `ENDPOINT_TYPE S3_TABLES`; that shortcut is
+for AWS S3 Tables endpoint and warehouse shapes.
+
+Run the repeatable DuckDB 1.5.5 smoke against an already running RustFS:
+
+```bash
+python3 scripts/table-catalog/duckdb_smoke.py \
+  --duckdb /path/to/duckdb \
+  --endpoint http://127.0.0.1:9000 \
+  --bucket rustfs-duckdb-smoke \
+  --namespace duckdb_smoke \
+  --table events \
+  --cleanup \
+  --rustfs-build rustfs-v1.0.0-rc.4 \
+  --git-sha "$(git rev-parse HEAD)" \
+  --catalog-backing object \
+  --live-evidence-output /tmp/rustfs-duckdb-live-evidence.json
+```
+
+The script requires the same PyIceberg, PyArrow, and boto3 dependencies as the
+PyIceberg smoke because it verifies both cross-engine directions. It creates an
+isolated namespace, keeps the final verified table at two rows for the shared
+evidence contract, and cleans all smoke tables only when `--cleanup` is set.
+It refuses to remove pre-existing suffixed smoke tables unless `--replace` is
+set explicitly. Cleanup preserves a namespace that existed before the run.
+
+The automated claim is limited to DuckDB 1.5.5, static S3 credentials, and the
+single-table scenarios exercised by this script. It does not claim DuckDB's AWS
+`S3_TABLES` shortcut, staged create, purge-on-drop, format v3, multi-table
+atomicity, or catalog-vended credential integration. The smoke verifies that
+DuckDB can run a two-table transaction with its multi-table commit endpoint
+disabled, but each table remains an independent RustFS commit.
+
 ## Spark Manual/Live Harness

 Spark validation should use the same RustFS endpoint and warehouse bucket as the
@@ -591,8 +656,9 @@ engines that are not run by default in RustFS CI:
 - Trino: catalog properties and a read-only `SELECT COUNT(*)` command for a
  table already created by PyIceberg or Spark. Trino write compatibility is not
  claimed.
- DuckDB: `httpfs` and `iceberg` SQL using an operator-supplied current Iceberg
-  metadata location. DuckDB write and commit compatibility are not claimed.
+- DuckDB: a legacy `httpfs` and `iceberg` read probe using an operator-supplied
+  current Iceberg metadata location. The separate `duckdb_smoke.py` entrypoint
+  owns the automated generic REST Catalog single-table read/write claim.
 - Databend: an S3 stage read probe for Parquet data files under the table
  warehouse. Databend Iceberg REST Catalog integration is not claimed.
 - Snowflake: an operator-adapted external volume/catalog integration SQL
@@ -0,0 +1,653 @@
+#!/usr/bin/env python3
+"""DuckDB Iceberg REST Catalog smoke test for RustFS S3 Tables."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import time
+import urllib.parse
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import engine_compatibility
+import pyiceberg_smoke
+
+
+DEFAULT_DUCKDB_VERSION = engine_compatibility.DEFAULT_DUCKDB_VERSION
+IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,47}$")
+
+
+@dataclass(frozen=True)
+class DuckDBExecution:
+    returncode: int
+    batches: list[list[dict[str, Any]]]
+    stdout: str
+    stderr: str
+
+
+@dataclass(frozen=True)
+class DuckDBSmokeResult:
+    client_version: str
+    metadata_location: str
+    row_count: int
+    cleanup_result: str
+    checks: dict[str, str]
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    run_id = str(int(time.time()))
+    parser = argparse.ArgumentParser(description="Run DuckDB Iceberg REST Catalog conformance against RustFS.")
+    parser.add_argument("--endpoint", default=os.getenv("RUSTFS_ENDPOINT", "http://127.0.0.1:9000"))
+    parser.add_argument("--access-key", default=os.getenv("RUSTFS_ACCESS_KEY", "rustfsadmin"))
+    parser.add_argument("--secret-key", default=os.getenv("RUSTFS_SECRET_KEY", "rustfsadmin"))
+    parser.add_argument("--region", default=os.getenv("RUSTFS_REGION", "us-east-1"))
+    parser.add_argument("--bucket", default=os.getenv("RUSTFS_TABLE_BUCKET", "rustfs-duckdb-smoke"))
+    parser.add_argument("--namespace", default=os.getenv("RUSTFS_TABLE_NAMESPACE", f"duckdb_smoke_{run_id}"))
+    parser.add_argument("--table", default=os.getenv("RUSTFS_TABLE_NAME", "events"))
+    parser.add_argument("--duckdb", default=os.getenv("DUCKDB_BIN", "duckdb"))
+    parser.add_argument("--duckdb-version", default=DEFAULT_DUCKDB_VERSION)
+    parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "60")))
+    parser.add_argument("--cleanup", action="store_true")
+    parser.add_argument("--replace", action="store_true", help="Drop existing smoke tables with matching identifiers first.")
+    parser.add_argument("--insecure", action="store_true")
+    parser.add_argument("--live-evidence-output")
+    parser.add_argument("--rustfs-build", default=os.getenv("RUSTFS_BUILD", "operator-recorded"))
+    parser.add_argument("--git-sha", default=os.getenv("RUSTFS_GIT_SHA", "operator-recorded"))
+    parser.add_argument("--catalog-backing", default=os.getenv("RUSTFS_TABLE_CATALOG_BACKING", "operator-recorded"))
+    parser.add_argument("--operator", default=os.getenv("USER", "operator-recorded"))
+    parser.add_argument("--run-timestamp-utc")
+    args = parser.parse_args(argv)
+    for label, value in [("namespace", args.namespace), ("table", args.table)]:
+        if not IDENTIFIER_RE.fullmatch(value):
+            parser.error(f"{label} must start with a letter and contain at most 48 ASCII letters, digits, or underscores")
+    return args
+
+
+def duckdb_path(value: str) -> str:
+    resolved = shutil.which(value)
+    if resolved is None:
+        raise RuntimeError(f"DuckDB executable was not found: {value}")
+    return resolved
+
+
+def duckdb_client_version(executable: str, timeout: float) -> str:
+    process = subprocess.run(
+        [executable, "-csv", "-noheader", "-c", "SELECT version();"],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+    )
+    if process.returncode != 0:
+        raise RuntimeError(f"DuckDB version probe failed: {process.stderr.strip()}")
+    version = process.stdout.strip().removeprefix("v")
+    if not version:
+        raise RuntimeError("DuckDB version probe returned an empty version")
+    return version
+
+
+def parse_duckdb_json(stdout: str) -> list[list[dict[str, Any]]]:
+    batches: list[list[dict[str, Any]]] = []
+    decoder = json.JSONDecoder()
+    offset = 0
+    while offset < len(stdout):
+        while offset < len(stdout) and stdout[offset].isspace():
+            offset += 1
+        if offset == len(stdout):
+            break
+        value, offset = decoder.raw_decode(stdout, offset)
+        if not isinstance(value, list) or any(not isinstance(row, dict) for row in value):
+            raise RuntimeError("DuckDB JSON output did not contain row objects")
+        batches.append(value)
+    return batches
+
+
+def run_duckdb(executable: str, sql: str, timeout: float) -> DuckDBExecution:
+    process = subprocess.run(
+        [executable, "-json", "-c", sql],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+    )
+    batches = parse_duckdb_json(process.stdout) if process.returncode == 0 else []
+    return DuckDBExecution(process.returncode, batches, process.stdout, process.stderr)
+
+
+def require_duckdb_success(execution: DuckDBExecution, label: str) -> None:
+    if execution.returncode != 0:
+        message = execution.stderr.strip() or execution.stdout.strip()
+        raise RuntimeError(f"DuckDB {label} failed: {message}")
+
+
+def require_duckdb_error(execution: DuckDBExecution, label: str, expected: str) -> None:
+    if execution.returncode == 0:
+        raise RuntimeError(f"DuckDB {label} unexpectedly succeeded")
+    message = f"{execution.stdout}\n{execution.stderr}"
+    if expected not in message:
+        raise RuntimeError(f"DuckDB {label} failed without expected error {expected!r}: {message.strip()}")
+
+
+def batches_with_column(execution: DuckDBExecution, column: str) -> list[list[dict[str, Any]]]:
+    return [batch for batch in execution.batches if batch and column in batch[0]]
+
+
+def table_name(base: str, suffix: str) -> str:
+    return f"{base}_{suffix}"
+
+
+def table_identifier(catalog: str, namespace: str, table: str) -> str:
+    return ".".join(
+        [
+            engine_compatibility.quote_double_identifier(catalog),
+            engine_compatibility.quote_double_identifier(namespace),
+            engine_compatibility.quote_double_identifier(table),
+        ]
+    )
+
+
+def profile_sql(
+    args: argparse.Namespace,
+    *,
+    catalog: str,
+    table: str,
+    rest_path: str = "/iceberg",
+    signing_name: str = "s3",
+) -> str:
+    return engine_compatibility.duckdb_rest_catalog_sql(
+        endpoint=args.endpoint,
+        warehouse=args.bucket,
+        access_key=args.access_key,
+        secret_key=args.secret_key,
+        region=args.region,
+        catalog_name=catalog,
+        namespace=args.namespace,
+        table=table,
+        rest_path=rest_path,
+        rest_signing_name=signing_name,
+    )
+
+
+def attach_sql(
+    args: argparse.Namespace,
+    *,
+    catalog: str,
+    rest_path: str,
+    signing_name: str,
+    compatibility_options: bool,
+    purge_requested: bool = False,
+) -> str:
+    options = [
+        "  TYPE iceberg",
+        f"  ENDPOINT {engine_compatibility.sql_string(f'{args.endpoint.rstrip('/')}{rest_path}')}",
+        "  AUTHORIZATION_TYPE 'sigv4'",
+        "  SECRET 'rustfs_s3'",
+        f"  SIGV4_REGION {engine_compatibility.sql_string(args.region)}",
+        f"  SIGV4_SERVICE {engine_compatibility.sql_string(signing_name)}",
+        "  ACCESS_DELEGATION_MODE 'none'",
+    ]
+    if compatibility_options:
+        options.extend(
+            [
+                "  STAGE_CREATE_TABLES false",
+                "  SKIP_CREATE_TABLE_METADATA_UPDATES true",
+                "  DISABLE_MULTI_TABLE_COMMIT true",
+                "  REMOVE_FILES_ON_DELETE false",
+                f"  PURGE_REQUESTED {'true' if purge_requested else 'false'}",
+                "  SUPPORT_NESTED_NAMESPACES false",
+            ]
+        )
+    rendered_options = ",\n".join(options)
+    return (
+        f"ATTACH {engine_compatibility.sql_string(args.bucket)} "
+        f"AS {engine_compatibility.quote_double_identifier(catalog)} (\n{rendered_options}\n);\n"
+    )
+
+
+def canonical_positive_sql(args: argparse.Namespace, seed_table: str, write_table: str, purge_table: str, drop_table: str) -> str:
+    catalog = "rustfs_duckdb"
+    namespace = ".".join(
+        [
+            engine_compatibility.quote_double_identifier(catalog),
+            engine_compatibility.quote_double_identifier(args.namespace),
+        ]
+    )
+    write_identifier = table_identifier(catalog, args.namespace, write_table)
+    purge_identifier = table_identifier(catalog, args.namespace, purge_table)
+    drop_identifier = table_identifier(catalog, args.namespace, drop_table)
+    return profile_sql(args, catalog=catalog, table=seed_table) + "\n".join(
+        [
+            f"CREATE SCHEMA IF NOT EXISTS {namespace};",
+            f"CREATE TABLE {write_identifier} (id BIGINT, payload VARCHAR);",
+            f"INSERT INTO {write_identifier} VALUES (10, 'ten'), (20, 'twenty');",
+            f"UPDATE {write_identifier} SET payload = 'TWENTY' WHERE id = 20;",
+            f"DELETE FROM {write_identifier} WHERE id = 10;",
+            f"ALTER TABLE {write_identifier} ADD COLUMN category VARCHAR;",
+            f"INSERT INTO {write_identifier} VALUES (30, 'thirty', 'new');",
+            f"MERGE INTO {write_identifier} AS target",
+            "USING (VALUES (20, 'twenty-merged', 'merged'), (40, 'forty', 'inserted')) AS source(id, payload, category)",
+            "ON target.id = source.id",
+            "WHEN MATCHED THEN UPDATE SET payload = source.payload, category = source.category",
+            "WHEN NOT MATCHED THEN INSERT (id, payload, category) VALUES (source.id, source.payload, source.category);",
+            f"DELETE FROM {write_identifier} WHERE id = 30;",
+            f"SELECT id, payload, category FROM {write_identifier} ORDER BY id;",
+            f"SELECT count(*) AS snapshot_count FROM iceberg_snapshots({write_identifier});",
+            f"CREATE TABLE {drop_identifier} (id BIGINT);",
+            f"INSERT INTO {drop_identifier} VALUES (1);",
+            f"DROP TABLE {drop_identifier};",
+            f"CREATE TABLE {purge_identifier} (id BIGINT);",
+            f"INSERT INTO {purge_identifier} VALUES (1);",
+            f"SELECT count(*) AS row_count FROM {write_identifier};",
+        ]
+    ) + "\n"
+
+
+def alias_sql(args: argparse.Namespace, write_table: str) -> str:
+    catalog = "rustfs_compat"
+    identifier = table_identifier(catalog, args.namespace, write_table)
+    return profile_sql(args, catalog=catalog, table=write_table, rest_path="/_iceberg", signing_name="s3tables") + "\n".join(
+        [
+            f"INSERT INTO {identifier} VALUES (50, 'fifty', 'compat');",
+            f"SELECT count(*) AS alias_row_count FROM {identifier};",
+            f"DELETE FROM {identifier} WHERE id = 50;",
+            f"SELECT count(*) AS alias_final_row_count FROM {identifier};",
+        ]
+    ) + "\n"
+
+
+def concurrent_insert_sql(args: argparse.Namespace, catalog: str, write_table: str, row_id: int) -> str:
+    identifier = table_identifier(catalog, args.namespace, write_table)
+    return profile_sql(args, catalog=catalog, table=write_table) + f"INSERT INTO {identifier} VALUES ({row_id}, 'writer-{row_id}', 'concurrent');\n"
+
+
+def multi_table_sql(args: argparse.Namespace, seed_table: str, write_table: str, purge_table: str) -> str:
+    catalog = "multi_table"
+    first = table_identifier(catalog, args.namespace, write_table)
+    second = table_identifier(catalog, args.namespace, purge_table)
+    return profile_sql(args, catalog=catalog, table=seed_table) + "\n".join(
+        [
+            "BEGIN TRANSACTION;",
+            f"INSERT INTO {first} VALUES (999, 'multi-a', 'non-atomic');",
+            f"INSERT INTO {second} VALUES (999);",
+            "COMMIT;",
+        ]
+    ) + "\n"
+
+
+def negative_sql(args: argparse.Namespace, *, kind: str, seed_table: str, write_table: str, purge_table: str) -> str:
+    bootstrap = f"bootstrap_{kind}"
+    sql = profile_sql(args, catalog=bootstrap, table=seed_table)
+    sql += f"DETACH {engine_compatibility.quote_double_identifier(bootstrap)};\n"
+    if kind == "stage-create":
+        catalog = "stage_default"
+        sql += attach_sql(
+            args,
+            catalog=catalog,
+            rest_path="/iceberg",
+            signing_name="s3",
+            compatibility_options=False,
+        )
+        sql += f"CREATE TABLE {table_identifier(catalog, args.namespace, table_name(args.table, 'stage'))} (id BIGINT);\n"
+        return sql
+    if kind == "purge":
+        catalog = "purge_requested"
+        sql += attach_sql(
+            args,
+            catalog=catalog,
+            rest_path="/iceberg",
+            signing_name="s3",
+            compatibility_options=True,
+            purge_requested=True,
+        )
+        sql += f"DROP TABLE {table_identifier(catalog, args.namespace, purge_table)};\n"
+        return sql
+    if kind == "format-v3":
+        catalog = "format_v3"
+        sql += attach_sql(
+            args,
+            catalog=catalog,
+            rest_path="/iceberg",
+            signing_name="s3",
+            compatibility_options=True,
+        )
+        identifier = table_identifier(catalog, args.namespace, table_name(args.table, "v3"))
+        sql += f"CREATE TABLE {identifier} (id BIGINT) WITH ('format-version' = '3');\n"
+        return sql
+    raise ValueError(f"unknown negative DuckDB smoke kind: {kind}")
+
+
+def pyiceberg_args(args: argparse.Namespace) -> argparse.Namespace:
+    return argparse.Namespace(
+        profile="rustfs",
+        endpoint=args.endpoint,
+        access_key=args.access_key,
+        secret_key=args.secret_key,
+        region=args.region,
+        bucket=args.bucket,
+        warehouse=None,
+        table_bucket=None,
+        account_id="000000000000",
+        warehouse_name=None,
+        catalog_uri=None,
+        namespace=args.namespace,
+        table=args.table,
+        catalog_name="rustfs_duckdb_pyiceberg",
+        rest_path="/iceberg",
+        rest_signing_name="s3",
+        require_vended_credentials=False,
+        timeout=args.timeout,
+        insecure=args.insecure,
+    )
+
+
+def prepare_smoke_tables(catalog: Any, namespace: str, tables: list[str], replace: bool) -> None:
+    existing = [table for table in tables if pyiceberg_smoke.table_exists(catalog, (namespace, table))]
+    if existing and not replace:
+        identifiers = ", ".join(f"{namespace}.{table}" for table in existing)
+        raise RuntimeError(f"DuckDB smoke tables already exist: {identifiers}; rerun with --replace to remove them")
+    for table in existing:
+        catalog.drop_table((namespace, table))
+
+
+def seed_pyiceberg_table(catalog: Any, args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps, table: str) -> None:
+    identifier = (args.namespace, table)
+    schema = deps.pyarrow.schema(
+        [
+            deps.pyarrow.field("id", deps.pyarrow.int64(), nullable=False),
+            deps.pyarrow.field("payload", deps.pyarrow.string(), nullable=False),
+        ]
+    )
+    created = catalog.create_table(identifier, schema=schema)
+    created.append(
+        deps.pyarrow.Table.from_pylist(
+            [{"id": 1, "payload": "alpha"}, {"id": 2, "payload": "beta"}],
+            schema=schema,
+        )
+    )
+
+
+def pyiceberg_rows(catalog: Any, namespace: str, table: str) -> list[dict[str, Any]]:
+    rows = catalog.load_table((namespace, table)).scan().to_arrow().to_pylist()
+    return sorted(rows, key=lambda row: row["id"])
+
+
+def run_concurrent_inserts(executable: str, args: argparse.Namespace, write_table: str) -> str:
+    probes = [("writer_a", 60), ("writer_b", 70)]
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        futures = [
+            executor.submit(run_duckdb, executable, concurrent_insert_sql(args, catalog, write_table, row_id), args.timeout)
+            for catalog, row_id in probes
+        ]
+    executions = [future.result() for future in futures]
+    retried = False
+    for (catalog, row_id), execution in zip(probes, executions, strict=True):
+        if execution.returncode == 0:
+            continue
+        error_text = f"{execution.stdout}\n{execution.stderr}".lower()
+        if not any(marker in error_text for marker in ["409", "conflict", "version token"]):
+            require_duckdb_success(execution, f"concurrent writer {row_id}")
+        retried = True
+        retry = run_duckdb(executable, concurrent_insert_sql(args, f"{catalog}_retry", write_table, row_id), args.timeout)
+        require_duckdb_success(retry, f"concurrent writer retry {row_id}")
+    return "passed-with-serial-retry" if retried else "passed-concurrently"
+
+
+def cleanup_tables(catalog: Any, namespace: str, tables: list[str], *, drop_namespace: bool) -> str:
+    cleanup_errors: list[str] = []
+    for table in tables:
+        try:
+            pyiceberg_smoke.drop_table_if_present(catalog, (namespace, table))
+        except Exception as error:
+            cleanup_errors.append(f"{table}: {error}")
+    if drop_namespace:
+        try:
+            catalog.drop_namespace(namespace)
+        except Exception as error:
+            cleanup_errors.append(f"namespace: {error}")
+    if cleanup_errors:
+        raise RuntimeError("DuckDB smoke cleanup failed: " + "; ".join(cleanup_errors))
+    return "dropped-tables-and-namespace" if drop_namespace else "dropped-tables-preserved-existing-namespace"
+
+
+def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> DuckDBSmokeResult:
+    executable = duckdb_path(args.duckdb)
+    client_version = duckdb_client_version(executable, args.timeout)
+    if client_version != args.duckdb_version:
+        raise RuntimeError(f"expected DuckDB {args.duckdb_version}, found {client_version}")
+
+    endpoint = pyiceberg_smoke.normalized_endpoint(args.endpoint)
+    pyiceberg_smoke.ensure_local_proxy_bypass(endpoint)
+    pyiceberg_smoke.ensure_aws_env(args.access_key, args.secret_key, args.region)
+    iceberg_args = pyiceberg_args(args)
+    pyiceberg_smoke.ensure_bucket(iceberg_args, deps)
+    pyiceberg_smoke.enable_table_bucket(iceberg_args, deps)
+
+    seed_table = table_name(args.table, "seed")
+    write_table = table_name(args.table, "write")
+    purge_table = table_name(args.table, "purge")
+    drop_table = table_name(args.table, "drop")
+    stage_table = table_name(args.table, "stage")
+    v3_table = table_name(args.table, "v3")
+    smoke_tables = [seed_table, write_table, purge_table, drop_table, stage_table, v3_table]
+    catalog = deps.load_catalog(iceberg_args.catalog_name, **pyiceberg_smoke.catalog_properties(iceberg_args))
+    pyiceberg_smoke.install_rustfs_rest_sigv4_adapter(catalog, iceberg_args, deps)
+    namespace_preexisting = bool(catalog.namespace_exists(args.namespace))
+    prepare_smoke_tables(catalog, args.namespace, smoke_tables, args.replace)
+    pyiceberg_smoke.ensure_namespace(catalog, args.namespace)
+    seed_pyiceberg_table(catalog, args, deps, seed_table)
+
+    checks: dict[str, str] = {}
+    cleanup_result = "not-requested"
+    metadata_location = "operator-recorded"
+    try:
+        positive = run_duckdb(
+            executable,
+            canonical_positive_sql(args, seed_table, write_table, purge_table, drop_table),
+            args.timeout,
+        )
+        require_duckdb_success(positive, "canonical REST catalog lifecycle")
+        row_count_batches = batches_with_column(positive, "row_count")
+        if len(row_count_batches) < 2 or row_count_batches[0][0]["row_count"] != 2 or row_count_batches[-1][0]["row_count"] != 2:
+            raise RuntimeError("DuckDB canonical REST catalog row counts did not remain at 2")
+        result_batches = batches_with_column(positive, "id")
+        expected_rows = [
+            {"id": 20, "payload": "twenty-merged", "category": "merged"},
+            {"id": 40, "payload": "forty", "category": "inserted"},
+        ]
+        if not result_batches or result_batches[-1] != expected_rows:
+            raise RuntimeError(f"DuckDB canonical DML returned unexpected rows: {result_batches[-1] if result_batches else []}")
+        snapshot_batches = batches_with_column(positive, "snapshot_count")
+        if not snapshot_batches or snapshot_batches[-1][0]["snapshot_count"] < 1:
+            raise RuntimeError("DuckDB snapshot metadata probe returned no snapshots")
+        if catalog.table_exists((args.namespace, drop_table)):
+            raise RuntimeError("DuckDB DROP TABLE did not remove the catalog entry")
+        checks["canonical_rest_catalog"] = "pass"
+        checks["single_table_ddl_dml"] = "pass"
+        checks["schema_evolution"] = "pass"
+        checks["snapshot_metadata"] = "pass"
+
+        if pyiceberg_rows(catalog, args.namespace, write_table) != expected_rows:
+            raise RuntimeError("PyIceberg did not observe DuckDB-created table rows")
+        checks["pyiceberg_cross_read"] = "pass"
+
+        alias = run_duckdb(executable, alias_sql(args, write_table), args.timeout)
+        require_duckdb_success(alias, "s3tables compatibility alias")
+        alias_counts = batches_with_column(alias, "alias_row_count")
+        alias_final_counts = batches_with_column(alias, "alias_final_row_count")
+        if not alias_counts or alias_counts[-1][0]["alias_row_count"] != 3:
+            raise RuntimeError("DuckDB compatibility alias insert did not produce row_count=3")
+        if not alias_final_counts or alias_final_counts[-1][0]["alias_final_row_count"] != 2:
+            raise RuntimeError("DuckDB compatibility alias cleanup did not restore row_count=2")
+        checks["s3tables_alias"] = "pass"
+
+        negatives = [
+            ("stage-create", "stage-create is not supported"),
+            ("purge", "purgeRequested=true is not supported"),
+            ("format-v3", "unsupported Iceberg table format-version: 3"),
+        ]
+        for kind, expected_error in negatives:
+            execution = run_duckdb(
+                executable,
+                negative_sql(
+                    args,
+                    kind=kind,
+                    seed_table=seed_table,
+                    write_table=write_table,
+                    purge_table=purge_table,
+                ),
+                args.timeout,
+            )
+            require_duckdb_error(execution, kind, expected_error)
+            checks[kind] = "failed-closed"
+        if catalog.table_exists((args.namespace, stage_table)) or catalog.table_exists((args.namespace, v3_table)):
+            raise RuntimeError("a failed DuckDB create probe left a catalog table behind")
+        if not catalog.table_exists((args.namespace, purge_table)):
+            raise RuntimeError("purgeRequested=true removed a table despite the expected failure")
+
+        multi_table = run_duckdb(
+            executable,
+            multi_table_sql(args, seed_table, write_table, purge_table),
+            args.timeout,
+        )
+        require_duckdb_success(multi_table, "multi-table endpoint-disabled mode")
+        if not any(row["id"] == 999 for row in pyiceberg_rows(catalog, args.namespace, write_table)):
+            raise RuntimeError("DuckDB multi-table endpoint-disabled mode did not commit the first table")
+        if not any(row["id"] == 999 for row in pyiceberg_rows(catalog, args.namespace, purge_table)):
+            raise RuntimeError("DuckDB multi-table endpoint-disabled mode did not commit the second table")
+        multi_cleanup_catalog = "cleanup_multi_table"
+        multi_cleanup = profile_sql(args, catalog=multi_cleanup_catalog, table=seed_table) + "\n".join(
+            [
+                f"DELETE FROM {table_identifier(multi_cleanup_catalog, args.namespace, write_table)} WHERE id = 999;",
+                f"DELETE FROM {table_identifier(multi_cleanup_catalog, args.namespace, purge_table)} WHERE id = 999;",
+            ]
+        )
+        require_duckdb_success(
+            run_duckdb(executable, multi_cleanup, args.timeout),
+            "multi-table endpoint-disabled cleanup",
+        )
+        checks["multi_table_endpoint_disabled"] = "pass-single-table-atomicity-only"
+
+        checks["concurrent_writers"] = run_concurrent_inserts(executable, args, write_table)
+        concurrent_rows = pyiceberg_rows(catalog, args.namespace, write_table)
+        if [row["id"] for row in concurrent_rows] != [20, 40, 60, 70]:
+            raise RuntimeError(f"concurrent DuckDB writers produced unexpected rows: {concurrent_rows}")
+        cleanup_catalog = "cleanup_concurrency"
+        cleanup_identifier = table_identifier(cleanup_catalog, args.namespace, write_table)
+        cleanup_sql = profile_sql(args, catalog=cleanup_catalog, table=write_table) + "\n".join(
+            [
+                f"DELETE FROM {cleanup_identifier} WHERE id IN (60, 70);",
+                f"SELECT count(*) AS final_row_count FROM {cleanup_identifier};",
+            ]
+        )
+        cleanup_execution = run_duckdb(executable, cleanup_sql, args.timeout)
+        require_duckdb_success(cleanup_execution, "concurrency cleanup")
+        final_batches = batches_with_column(cleanup_execution, "final_row_count")
+        if not final_batches or final_batches[-1][0]["final_row_count"] != 2:
+            raise RuntimeError("DuckDB concurrency cleanup did not restore row_count=2")
+
+        final_table = catalog.load_table((args.namespace, write_table))
+        final_rows = sorted(final_table.scan().to_arrow().to_pylist(), key=lambda row: row["id"])
+        if final_rows != expected_rows:
+            raise RuntimeError(f"final PyIceberg cross-read returned unexpected rows: {final_rows}")
+        metadata_location = pyiceberg_smoke.table_metadata_location(final_table) or "operator-recorded"
+        if metadata_location == "operator-recorded":
+            response = pyiceberg_smoke.signed_rest_request(
+                argparse.Namespace(**{**vars(iceberg_args), "table": write_table}),
+                deps,
+                "GET",
+                f"/iceberg/v1/{urllib.parse.quote(args.bucket, safe='')}/namespaces/"
+                f"{urllib.parse.quote(args.namespace, safe='')}/tables/{urllib.parse.quote(write_table, safe='')}",
+            )
+            metadata_location = response.get("metadata-location", "operator-recorded")
+        if metadata_location == "operator-recorded":
+            raise RuntimeError("DuckDB smoke could not resolve the final metadata location")
+        metadata_scan = run_duckdb(
+            executable,
+            engine_compatibility.duckdb_sql_probe(
+                endpoint=args.endpoint,
+                access_key=args.access_key,
+                secret_key=args.secret_key,
+                region=args.region,
+                metadata_location=metadata_location,
+            ),
+            args.timeout,
+        )
+        require_duckdb_success(metadata_scan, "metadata-location scan")
+        metadata_counts = batches_with_column(metadata_scan, "row_count")
+        if not metadata_counts or metadata_counts[-1][0]["row_count"] != 2:
+            raise RuntimeError("DuckDB metadata-location scan did not return row_count=2")
+        checks["metadata_location_scan"] = "pass"
+    finally:
+        if args.cleanup:
+            cleanup_result = cleanup_tables(
+                catalog,
+                args.namespace,
+                smoke_tables,
+                drop_namespace=not namespace_preexisting,
+            )
+
+    return DuckDBSmokeResult(client_version, metadata_location, 2, cleanup_result, checks)
+
+
+def current_utc_timestamp() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def write_live_evidence(args: argparse.Namespace, result: DuckDBSmokeResult) -> None:
+    if not args.live_evidence_output:
+        return
+    command = pyiceberg_smoke.redacted_command(sys.argv)
+    record = engine_compatibility.live_conformance_evidence_record(
+        client_name="DuckDB Iceberg",
+        client_version=result.client_version,
+        scenario="rest-catalog-single-table-read-write-cross-engine-negative-boundaries",
+        rustfs_build=args.rustfs_build,
+        git_sha=args.git_sha,
+        catalog_backing=args.catalog_backing,
+        endpoint=args.endpoint,
+        warehouse=args.bucket,
+        rest_path="/iceberg",
+        namespace=args.namespace,
+        table=table_name(args.table, "write"),
+        metadata_location=result.metadata_location,
+        run_timestamp_utc=args.run_timestamp_utc or current_utc_timestamp(),
+        operator=args.operator,
+        expected_status="pass",
+        observed_status="pass",
+        row_count=result.row_count,
+        cleanup_result=result.cleanup_result,
+        claim="automated-rest-catalog-smoke",
+        command=command,
+    )
+    document = {
+        "live_conformance_evidence": record,
+        "checks": result.checks,
+        "validation": engine_compatibility.validate_live_conformance_evidence(record),
+    }
+    Path(args.live_evidence_output).write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        deps = pyiceberg_smoke.load_runtime_deps()
+        result = run_smoke(args, deps)
+        write_live_evidence(args, result)
+        print(json.dumps({"status": "pass", "row_count": result.row_count, "checks": result.checks}, sort_keys=True))
+        return 0
+    except Exception as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -17,7 +17,7 @@ DEFAULT_SPARK_VERSION = "3.5.4"
 DEFAULT_ICEBERG_VERSION = "1.7.1"
 DEFAULT_SCALA_VERSION = "2.12"
 DEFAULT_TRINO_VERSION = "477"
-DEFAULT_DUCKDB_VERSION = "1.3.2"
+DEFAULT_DUCKDB_VERSION = "1.5.5"
 DEFAULT_SNOWFLAKE_CLIENT_VERSION = "operator-recorded"
 DEFAULT_DATABEND_VERSION = "operator-recorded"
 DEFAULT_TRINO_SERVER = "http://127.0.0.1:8080"
@@ -53,7 +53,7 @@ LIVE_EVIDENCE_ALLOWED_CLAIMS = OrderedDict(
        ("PyIceberg", ["automated-smoke"]),
        ("Spark Iceberg REST catalog", ["manual-live-verified"]),
        ("Trino Iceberg REST catalog", ["manual-live-read-verified"]),
-        ("DuckDB Iceberg", ["manual-live-read-verified"]),
+        ("DuckDB Iceberg", ["manual-live-read-verified", "automated-rest-catalog-smoke"]),
        ("Databend", ["manual-live-s3-stage-verified"]),
        ("Snowflake Open Catalog / Iceberg integrations", ["reference-only"]),
    ]
@@ -155,12 +155,15 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]:
        },
        {
            "client": "DuckDB Iceberg",
-            "status": "manual-live-read-probe",
-            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
+            "status": "automated-smoke",
+            "entrypoint": "scripts/table-catalog/duckdb_smoke.py",
            "scenarios": [
-                scenario("metadata-read", "manual-live-probe", "read a supplied Iceberg metadata location through DuckDB iceberg_scan"),
-                scenario("read-table", "manual-live-probe", "read-path verification only"),
-                scenario("write-table", "not-claimed", "DuckDB write/commit compatibility is not claimed"),
+                scenario("metadata-read", "automated", "read the final metadata location through DuckDB iceberg_scan"),
+                scenario("catalog-attach", "automated", "attach `/iceberg` with s3 signing and `/_iceberg` with s3tables signing"),
+                scenario("read-table", "automated", "read a PyIceberg-created table through the attached catalog"),
+                scenario("write-table", "automated", "exercise single-table DDL, DML, schema evolution, snapshots, and PyIceberg cross-read"),
+                scenario("unsupported-boundaries", "automated", "verify staged create, purge, and format v3 fail closed"),
+                scenario("multi-table-mode", "automated", "verify DuckDB can avoid the multi-table commit endpoint without claiming cross-table atomicity"),
            ],
        },
        {
@@ -489,8 +492,66 @@ def duckdb_sql_probe(
    return "\n".join(statements) + "\n"


-def duckdb_command() -> str:
-    return shell_join(["duckdb", "-c", ".read /tmp/rustfs-s3tables-duckdb-read.sql"])
+def duckdb_rest_catalog_sql(
+    *,
+    endpoint: str,
+    warehouse: str,
+    access_key: str,
+    secret_key: str,
+    region: str,
+    catalog_name: str,
+    namespace: str,
+    table: str,
+    rest_path: str,
+    rest_signing_name: str,
+) -> str:
+    parsed = re.match(r"^(https?)://(.+)$", normalized_endpoint(endpoint))
+    if not parsed:
+        raise ValueError("DuckDB REST catalog endpoint must include http:// or https://")
+    scheme, endpoint_without_scheme = parsed.groups()
+    rest_path = normalized_rest_path(rest_path)
+    catalog_identifier = quote_double_identifier(catalog_name)
+    table_identifier = ".".join(
+        [catalog_identifier, quote_double_identifier(namespace), quote_double_identifier(table)]
+    )
+    statements = [
+        "INSTALL httpfs;",
+        "LOAD httpfs;",
+        "INSTALL iceberg;",
+        "LOAD iceberg;",
+        "CREATE OR REPLACE SECRET rustfs_s3 (",
+        "  TYPE s3,",
+        "  PROVIDER config,",
+        f"  KEY_ID {sql_string(access_key)},",
+        f"  SECRET {sql_string(secret_key)},",
+        f"  REGION {sql_string(region)},",
+        f"  ENDPOINT {sql_string(endpoint_without_scheme)},",
+        "  URL_STYLE 'path',",
+        f"  USE_SSL {'true' if scheme == 'https' else 'false'},",
+        f"  SCOPE {sql_string(f's3://{warehouse}')}",
+        ");",
+        f"ATTACH {sql_string(warehouse)} AS {catalog_identifier} (",
+        "  TYPE iceberg,",
+        f"  ENDPOINT {sql_string(f'{normalized_endpoint(endpoint)}{rest_path}')},",
+        "  AUTHORIZATION_TYPE 'sigv4',",
+        "  SECRET 'rustfs_s3',",
+        f"  SIGV4_REGION {sql_string(region)},",
+        f"  SIGV4_SERVICE {sql_string(rest_signing_name)},",
+        "  ACCESS_DELEGATION_MODE 'none',",
+        "  STAGE_CREATE_TABLES false,",
+        "  SKIP_CREATE_TABLE_METADATA_UPDATES true,",
+        "  DISABLE_MULTI_TABLE_COMMIT true,",
+        "  REMOVE_FILES_ON_DELETE false,",
+        "  PURGE_REQUESTED false,",
+        "  SUPPORT_NESTED_NAMESPACES false",
+        ");",
+        f"SELECT COUNT(*) AS row_count FROM {table_identifier};",
+    ]
+    return "\n".join(statements) + "\n"
+
+
+def duckdb_command(*, sql_file: str = "/tmp/rustfs-s3tables-duckdb-read.sql") -> str:
+    return shell_join(["duckdb", "-c", f".read {sql_file}"])


 def snowflake_sql_template(*, endpoint: str, warehouse: str, rest_path: str, namespace: str, table: str) -> str:
@@ -623,11 +684,11 @@ def live_conformance_evidence(
                    OrderedDict(
                        [
                            ("client", "DuckDB Iceberg"),
-                            ("scenario", "iceberg-scan-current-metadata-location"),
+                            ("scenario", "rest-catalog-single-table-read-write-cross-engine-negative-boundaries"),
                            ("expected_status", "pass"),
                            ("expected_row_count", 2),
-                            ("claim_after_pass", "manual-live-read-verified"),
-                            ("write_claim_after_pass", "not-claimed"),
+                            ("claim_after_pass", "automated-rest-catalog-smoke"),
+                            ("write_claim_after_pass", "single-table-automated-smoke"),
                        ]
                    ),
                    OrderedDict(
@@ -653,9 +714,9 @@ def live_conformance_evidence(
            (
                "promotion_rules",
                [
-                    "Keep PyIceberg as the only automated claim unless the run is executed by CI or a repeatable operator job.",
+                    "Keep PyIceberg and DuckDB automated claims tied to their repeatable smoke entrypoints and recorded client versions.",
                    "Promote Spark only to manual-live-verified when the exact RustFS build, Spark version, Iceberg version, SQL output, and row_count are recorded.",
-                    "Do not promote Trino or DuckDB write compatibility from read probes; write compatibility remains not-claimed.",
+                    "Keep Trino write compatibility not-claimed after its read probe and do not broaden DuckDB beyond the automated single-table scenarios.",
                    "Do not promote Snowflake or vendor catalog interoperability from a generated template without a repeatable live run.",
                    "Treat manual-live failures as compatibility findings and keep the previous public claim boundary.",
                ],
@@ -1425,6 +1486,18 @@ def live_conformance_harness(
        region=region,
        metadata_location=metadata_location,
    )
+    duckdb_rest_sql = duckdb_rest_catalog_sql(
+        endpoint=endpoint,
+        warehouse=warehouse,
+        access_key=access_key,
+        secret_key=secret_key,
+        region=region,
+        catalog_name=catalog_name,
+        namespace=namespace,
+        table=table,
+        rest_path=rest_path,
+        rest_signing_name=rest_signing_name,
+    )
    snowflake_sql = snowflake_sql_template(
        endpoint=endpoint,
        warehouse=warehouse,
@@ -1553,14 +1626,25 @@ def live_conformance_harness(
                    OrderedDict(
                        [
                            ("name", "DuckDB Iceberg"),
-                            ("status", "manual-live-read-probe"),
+                            ("status", "automated-smoke"),
                            ("version", duckdb_version),
                            ("metadata_location", metadata_location),
                            ("sql_file", "/tmp/rustfs-s3tables-duckdb-read.sql"),
                            ("sql", duckdb_sql),
                            ("command", duckdb_command()),
                            ("expected", "iceberg_scan returns row_count=2 when metadata_location points at the current Iceberg metadata JSON"),
-                            ("write_compatibility", "not-claimed"),
+                            ("rest_catalog_sql_file", "/tmp/rustfs-s3tables-duckdb-rest.sql"),
+                            ("rest_catalog_sql", duckdb_rest_sql),
+                            (
+                                "rest_catalog_command",
+                                duckdb_command(sql_file="/tmp/rustfs-s3tables-duckdb-rest.sql"),
+                            ),
+                            (
+                                "rest_catalog_expected",
+                                "generic Iceberg REST ATTACH returns row_count=2 for an existing RustFS table",
+                            ),
+                            ("rest_catalog_write_compatibility", "single-table-automated-smoke"),
+                            ("write_compatibility", "single-table-automated-smoke"),
                        ]
                    ),
                    OrderedDict(
@@ -1627,6 +1711,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser.add_argument("--print-operations-guide", action="store_true")
    parser.add_argument("--print-spark-config", action="store_true")
    parser.add_argument("--print-spark-sql", action="store_true")
+    parser.add_argument("--print-duckdb-rest-sql", action="store_true")
    return parser.parse_args(argv)


@@ -1734,6 +1819,24 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None:
        else:
            output.write(sql)
        printed = True
+    if args.print_duckdb_rest_sql:
+        sql = duckdb_rest_catalog_sql(
+            endpoint=args.endpoint,
+            warehouse=args.warehouse,
+            access_key=args.access_key,
+            secret_key=args.secret_key,
+            region=args.region,
+            catalog_name=args.catalog_name,
+            namespace=args.namespace,
+            table=args.table,
+            rest_path=args.rest_path or "/iceberg",
+            rest_signing_name=args.rest_signing_name or "s3",
+        )
+        if output is None:
+            print(sql, end="")
+        else:
+            output.write(sql)
+        printed = True
    if not printed:
        print_json({"engine_compatibility": engine_compatibility_matrix()}, output)

@@ -167,9 +167,9 @@ CLIENT_MATRIX: list[dict[str, str]] = [
    },
    {
        "client": "DuckDB Iceberg",
-        "status": "manual-live-read-probe",
-        "coverage": "generated httpfs/iceberg SQL using an operator-supplied current metadata location; write/commit is not claimed",
-        "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
+        "status": "automated-smoke",
+        "coverage": "metadata-location read plus generic REST catalog single-table DDL, DML, schema evolution, snapshots, canonical and compatibility signing, negative boundaries, endpoint-disabled multi-table mode, concurrent writers, and PyIceberg cross-read",
+        "entrypoint": "scripts/table-catalog/duckdb_smoke.py",
    },
    {
        "client": "Databend",
@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""Unit tests for the RustFS DuckDB REST Catalog smoke helper."""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import duckdb_smoke
+
+
+class DuckDBSmokeTest(unittest.TestCase):
+    def args(self) -> argparse.Namespace:
+        return argparse.Namespace(
+            endpoint="http://127.0.0.1:9000",
+            access_key="rustfsadmin",
+            secret_key="rustfsadmin",
+            region="us-east-1",
+            bucket="rustfs-duckdb-smoke",
+            namespace="duckdb_smoke",
+            table="events",
+            duckdb="duckdb",
+            duckdb_version="1.5.5",
+            timeout=60.0,
+            cleanup=True,
+            replace=False,
+            insecure=False,
+            live_evidence_output=None,
+            rustfs_build="rustfs-test",
+            git_sha="abc123",
+            catalog_backing="object",
+            operator="test-operator",
+            run_timestamp_utc="2026-08-27T00:00:00Z",
+        )
+
+    def test_parse_args_rejects_unsafe_identifiers(self) -> None:
+        with contextlib.redirect_stderr(io.StringIO()):
+            with self.assertRaises(SystemExit):
+                duckdb_smoke.parse_args(["--namespace", "bad-name"])
+
+    def test_duckdb_client_version_removes_v_prefix(self) -> None:
+        process = SimpleNamespace(returncode=0, stdout="v1.5.5\n", stderr="")
+        with mock.patch.object(duckdb_smoke.subprocess, "run", return_value=process):
+            self.assertEqual(duckdb_smoke.duckdb_client_version("duckdb", 10), "1.5.5")
+
+    def test_parse_duckdb_json_accepts_multiple_query_batches(self) -> None:
+        batches = duckdb_smoke.parse_duckdb_json(
+            '[{"row_count":2}]\n[{"id":20},\n{"id":40}]\n'
+        )
+
+        self.assertEqual(batches[0][0]["row_count"], 2)
+        self.assertEqual([row["id"] for row in batches[1]], [20, 40])
+
+    def test_run_duckdb_does_not_parse_json_for_failed_process(self) -> None:
+        process = SimpleNamespace(returncode=1, stdout="not-json", stderr="expected failure")
+        with mock.patch.object(duckdb_smoke.subprocess, "run", return_value=process):
+            execution = duckdb_smoke.run_duckdb("duckdb", "SELECT 1", 10)
+
+        self.assertEqual(execution.returncode, 1)
+        self.assertEqual(execution.batches, [])
+
+    def test_concurrent_retry_markers_do_not_accept_generic_commit_errors(self) -> None:
+        args = self.args()
+        generic_failure = duckdb_smoke.DuckDBExecution(1, [], "", "commit failed: permission denied")
+        with mock.patch.object(duckdb_smoke, "run_duckdb", return_value=generic_failure):
+            with self.assertRaisesRegex(RuntimeError, "permission denied"):
+                duckdb_smoke.run_concurrent_inserts("duckdb", args, "events_write")
+
+    def test_canonical_sql_covers_single_table_lifecycle(self) -> None:
+        sql = duckdb_smoke.canonical_positive_sql(
+            self.args(),
+            "events_seed",
+            "events_write",
+            "events_purge",
+            "events_drop",
+        )
+
+        self.assertIn("STAGE_CREATE_TABLES false", sql)
+        self.assertIn("SKIP_CREATE_TABLE_METADATA_UPDATES true", sql)
+        self.assertIn("CREATE TABLE", sql)
+        self.assertIn("INSERT INTO", sql)
+        self.assertIn("UPDATE", sql)
+        self.assertIn("DELETE FROM", sql)
+        self.assertIn("MERGE INTO", sql)
+        self.assertIn("ALTER TABLE", sql)
+        self.assertIn("iceberg_snapshots", sql)
+        self.assertNotIn("DROP TABLE IF EXISTS", sql)
+        self.assertIn('DROP TABLE "rustfs_duckdb"."duckdb_smoke"."events_drop"', sql)
+
+    def test_prepare_smoke_tables_refuses_existing_identifiers_without_replace(self) -> None:
+        catalog = mock.Mock()
+        catalog.table_exists.side_effect = lambda identifier: identifier[1] == "events_write"
+
+        with self.assertRaisesRegex(RuntimeError, "duckdb_smoke.events_write"):
+            duckdb_smoke.prepare_smoke_tables(
+                catalog,
+                "duckdb_smoke",
+                ["events_seed", "events_write"],
+                replace=False,
+            )
+
+        catalog.drop_table.assert_not_called()
+
+    def test_prepare_smoke_tables_replaces_only_existing_identifiers_when_requested(self) -> None:
+        catalog = mock.Mock()
+        catalog.table_exists.side_effect = lambda identifier: identifier[1] == "events_write"
+
+        duckdb_smoke.prepare_smoke_tables(
+            catalog,
+            "duckdb_smoke",
+            ["events_seed", "events_write"],
+            replace=True,
+        )
+
+        catalog.drop_table.assert_called_once_with(("duckdb_smoke", "events_write"))
+
+    def test_cleanup_preserves_a_preexisting_namespace(self) -> None:
+        catalog = mock.Mock()
+        catalog.table_exists.return_value = False
+
+        result = duckdb_smoke.cleanup_tables(
+            catalog,
+            "duckdb_smoke",
+            ["events_seed", "events_write"],
+            drop_namespace=False,
+        )
+
+        self.assertEqual(result, "dropped-tables-preserved-existing-namespace")
+        catalog.drop_namespace.assert_not_called()
+
+    def test_alias_sql_uses_s3tables_signing(self) -> None:
+        sql = duckdb_smoke.alias_sql(self.args(), "events_write")
+
+        self.assertIn("ENDPOINT 'http://127.0.0.1:9000/_iceberg'", sql)
+        self.assertIn("SIGV4_SERVICE 's3tables'", sql)
+        self.assertIn("INSERT INTO", sql)
+        self.assertIn("alias_final_row_count", sql)
+
+    def test_boundary_sql_records_required_compatibility_options(self) -> None:
+        args = self.args()
+        stage_sql = duckdb_smoke.negative_sql(
+            args,
+            kind="stage-create",
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        stage_attach = stage_sql.split('DETACH "bootstrap_stage-create";', 1)[1]
+        self.assertNotIn("STAGE_CREATE_TABLES false", stage_attach)
+        self.assertIn("CREATE TABLE", stage_attach)
+
+        purge_sql = duckdb_smoke.negative_sql(
+            args,
+            kind="purge",
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        self.assertIn("PURGE_REQUESTED true", purge_sql)
+        self.assertIn('DROP TABLE "purge_requested"."duckdb_smoke"."events_purge"', purge_sql)
+
+        v3_sql = duckdb_smoke.negative_sql(
+            args,
+            kind="format-v3",
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        self.assertIn("'format-version' = '3'", v3_sql)
+
+        multi_sql = duckdb_smoke.multi_table_sql(
+            args,
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        self.assertIn("DISABLE_MULTI_TABLE_COMMIT true", multi_sql)
+        self.assertIn("BEGIN TRANSACTION", multi_sql)
+        self.assertIn("'non-atomic'", multi_sql)
+
+    def test_pyiceberg_args_use_canonical_catalog(self) -> None:
+        args = duckdb_smoke.pyiceberg_args(self.args())
+
+        self.assertEqual(args.rest_path, "/iceberg")
+        self.assertEqual(args.rest_signing_name, "s3")
+        self.assertEqual(args.bucket, "rustfs-duckdb-smoke")
+
+    def test_live_evidence_records_automated_duckdb_claim(self) -> None:
+        args = self.args()
+        result = duckdb_smoke.DuckDBSmokeResult(
+            client_version="1.5.5",
+            metadata_location="s3://rustfs-duckdb-smoke/metadata/00001.json",
+            row_count=2,
+            cleanup_result="dropped-tables-and-namespace",
+            checks={"canonical_rest_catalog": "pass"},
+        )
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output = Path(temp_dir) / "evidence.json"
+            args.live_evidence_output = str(output)
+            with mock.patch.object(duckdb_smoke.sys, "argv", ["duckdb_smoke.py", "--secret-key", "secret"]):
+                duckdb_smoke.write_live_evidence(args, result)
+            document = json.loads(output.read_text(encoding="utf-8"))
+
+        evidence = document["live_conformance_evidence"]
+        self.assertEqual(evidence["client_name"], "DuckDB Iceberg")
+        self.assertEqual(evidence["claim"], "automated-rest-catalog-smoke")
+        self.assertIn("--secret-key '<redacted>'", evidence["command"])
+        self.assertNotIn("--secret-key secret", evidence["command"])
+        self.assertEqual(document["validation"]["status"], "accepted")
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -41,6 +41,15 @@ class EngineCompatibilityTest(unittest.TestCase):
        self.assertEqual(trino["status"], "manual-live-read-probe")
        self.assertContainsScenario(trino, "catalog-load", "manual-live-probe")

+        duckdb = by_client["DuckDB Iceberg"]
+        self.assertEqual(duckdb["status"], "automated-smoke")
+        self.assertEqual(duckdb["entrypoint"], "scripts/table-catalog/duckdb_smoke.py")
+        self.assertContainsScenario(duckdb, "metadata-read", "automated")
+        self.assertContainsScenario(duckdb, "catalog-attach", "automated")
+        self.assertContainsScenario(duckdb, "write-table", "automated")
+        self.assertContainsScenario(duckdb, "unsupported-boundaries", "automated")
+        self.assertContainsScenario(duckdb, "multi-table-mode", "automated")
+
    def test_spark_config_uses_rustfs_rest_catalog_and_s3fileio(self) -> None:
        config = engine_compatibility.spark_catalog_config(
            endpoint="http://127.0.0.1:9000",
@@ -158,6 +167,70 @@ class EngineCompatibilityTest(unittest.TestCase):
                table="orders",
            )

+    def test_duckdb_rest_catalog_sql_uses_rustfs_compatibility_options(self) -> None:
+        sql = engine_compatibility.duckdb_rest_catalog_sql(
+            endpoint="http://127.0.0.1:9000",
+            warehouse="rustfs-s3table-smoke",
+            access_key="rustfsadmin",
+            secret_key="rustfsadmin",
+            region="us-east-1",
+            catalog_name="rustfs",
+            namespace="smoke",
+            table="events",
+            rest_path="/iceberg",
+            rest_signing_name="s3",
+        )
+
+        self.assertIn("CREATE OR REPLACE SECRET rustfs_s3", sql)
+        self.assertIn("ENDPOINT '127.0.0.1:9000'", sql)
+        self.assertIn("SCOPE 's3://rustfs-s3table-smoke'", sql)
+        self.assertIn("ATTACH 'rustfs-s3table-smoke' AS \"rustfs\"", sql)
+        self.assertIn("ENDPOINT 'http://127.0.0.1:9000/iceberg'", sql)
+        self.assertIn("SIGV4_SERVICE 's3'", sql)
+        self.assertIn("STAGE_CREATE_TABLES false", sql)
+        self.assertIn("SKIP_CREATE_TABLE_METADATA_UPDATES true", sql)
+        self.assertIn("DISABLE_MULTI_TABLE_COMMIT true", sql)
+        self.assertIn("REMOVE_FILES_ON_DELETE false", sql)
+        self.assertIn("PURGE_REQUESTED false", sql)
+        self.assertIn('FROM "rustfs"."smoke"."events"', sql)
+        self.assertNotIn("ENDPOINT_TYPE", sql)
+
+    def test_duckdb_rest_catalog_sql_supports_s3tables_alias(self) -> None:
+        sql = engine_compatibility.duckdb_rest_catalog_sql(
+            endpoint="https://rustfs.example",
+            warehouse="analytics",
+            access_key="access'key",
+            secret_key="secret'key",
+            region="us-east-1",
+            catalog_name="rustfs_compat",
+            namespace="smoke",
+            table="events",
+            rest_path="/_iceberg",
+            rest_signing_name="s3tables",
+        )
+
+        self.assertIn("USE_SSL true", sql)
+        self.assertIn("ENDPOINT 'rustfs.example'", sql)
+        self.assertIn("ENDPOINT 'https://rustfs.example/_iceberg'", sql)
+        self.assertIn("SIGV4_SERVICE 's3tables'", sql)
+        self.assertIn("KEY_ID 'access''key'", sql)
+        self.assertIn("SECRET 'secret''key'", sql)
+
+    def test_duckdb_rest_catalog_sql_rejects_endpoint_without_scheme(self) -> None:
+        with self.assertRaisesRegex(ValueError, "must include http:// or https://"):
+            engine_compatibility.duckdb_rest_catalog_sql(
+                endpoint="127.0.0.1:9000",
+                warehouse="analytics",
+                access_key="rustfsadmin",
+                secret_key="rustfsadmin",
+                region="us-east-1",
+                catalog_name="rustfs",
+                namespace="smoke",
+                table="events",
+                rest_path="/iceberg",
+                rest_signing_name="s3",
+            )
+
    def test_cli_prints_machine_readable_engine_matrix(self) -> None:
        payload = engine_compatibility.cli_json(["--print-engine-matrix"])
        document = json.loads(payload)
@@ -308,6 +381,27 @@ class EngineCompatibilityTest(unittest.TestCase):
        self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "http://127.0.0.1:9000/_iceberg")
        self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables")

+    def test_cli_prints_duckdb_rest_catalog_sql(self) -> None:
+        sql = engine_compatibility.cli_json(
+            [
+                "--print-duckdb-rest-sql",
+                "--endpoint",
+                "http://127.0.0.1:9000",
+                "--warehouse",
+                "analytics",
+                "--catalog-name",
+                "rustfs_compat",
+                "--rest-path",
+                "/_iceberg",
+                "--rest-signing-name",
+                "s3tables",
+            ]
+        )
+
+        self.assertIn("ATTACH 'analytics' AS \"rustfs_compat\"", sql)
+        self.assertIn("ENDPOINT 'http://127.0.0.1:9000/_iceberg'", sql)
+        self.assertIn("SIGV4_SERVICE 's3tables'", sql)
+
    def test_live_conformance_harness_pins_clients_and_records_commands(self) -> None:
        harness = engine_compatibility.live_conformance_harness(
            endpoint="http://127.0.0.1:9000",
@@ -359,11 +453,16 @@ class EngineCompatibilityTest(unittest.TestCase):
        self.assertEqual(trino["write_compatibility"], "not-claimed")

        duckdb = by_client["DuckDB Iceberg"]
-        self.assertEqual(duckdb["status"], "manual-live-read-probe")
+        self.assertEqual(duckdb["status"], "automated-smoke")
+        self.assertEqual(duckdb["version"], "1.5.5")
        self.assertIn("LOAD httpfs", duckdb["sql"])
        self.assertIn("LOAD iceberg", duckdb["sql"])
        self.assertIn("iceberg_scan", duckdb["sql"])
-        self.assertEqual(duckdb["write_compatibility"], "not-claimed")
+        self.assertIn("ATTACH 'rustfs-s3table-smoke'", duckdb["rest_catalog_sql"])
+        self.assertIn("STAGE_CREATE_TABLES false", duckdb["rest_catalog_sql"])
+        self.assertIn("SKIP_CREATE_TABLE_METADATA_UPDATES true", duckdb["rest_catalog_sql"])
+        self.assertEqual(duckdb["rest_catalog_write_compatibility"], "single-table-automated-smoke")
+        self.assertEqual(duckdb["write_compatibility"], "single-table-automated-smoke")

        snowflake = by_client["Snowflake Open Catalog / Iceberg integrations"]
        self.assertEqual(snowflake["status"], "manual-reference-probe")
@@ -407,7 +506,8 @@ class EngineCompatibilityTest(unittest.TestCase):
        self.assertEqual(table_by_client["PyIceberg"]["claim_after_pass"], "automated-smoke")
        self.assertEqual(table_by_client["Spark Iceberg REST catalog"]["claim_after_pass"], "manual-live-verified")
        self.assertEqual(table_by_client["Trino Iceberg REST catalog"]["write_claim_after_pass"], "not-claimed")
-        self.assertEqual(table_by_client["DuckDB Iceberg"]["write_claim_after_pass"], "not-claimed")
+        self.assertEqual(table_by_client["DuckDB Iceberg"]["claim_after_pass"], "automated-rest-catalog-smoke")
+        self.assertEqual(table_by_client["DuckDB Iceberg"]["write_claim_after_pass"], "single-table-automated-smoke")
        self.assertIn("manual-live", " ".join(evidence["promotion_rules"]))
        self.assertIn("not-claimed", " ".join(evidence["promotion_rules"]))

@@ -496,6 +596,10 @@ class EngineCompatibilityTest(unittest.TestCase):
        self.assertIn("metadata_location", schema["required_fields"])
        self.assertIn("claim", schema["required_fields"])
        self.assertEqual(schema["claim_promotion"]["Trino Iceberg REST catalog"], ["manual-live-read-verified"])
+        self.assertEqual(
+            schema["claim_promotion"]["DuckDB Iceberg"],
+            ["manual-live-read-verified", "automated-rest-catalog-smoke"],
+        )

    def test_production_operations_guide_covers_release_boundaries(self) -> None:
        guide = engine_compatibility.production_operations_guide(
@@ -25,14 +25,17 @@ All status checks talk to the RustFS admin API directly (SigV4-signed,
 5. Starts cluster heal: `POST /rustfs/admin/v3/heal/` with body
   `{"recursive":true}` (retried, returns a `clientToken`).
 6. Monitors the heal task via `POST /rustfs/admin/v3/heal/?clientToken=<token>`
-   until the summary is a terminal success (`finished`/`completed`),
-   `objects_failed == 0`, **and** the outage node's disk usage reaches
-   `HEAL_TARGET_GB` (default 40 GiB).
-7. Result analysis: heal stats (scanned/healed/failed), per-node disk usage,
+   until the server verdict is a terminal success (`finished`/`completed`) with
+   `objects_failed == 0`.
+7. Result analysis: heal stats (scanned/healed/failed), an **S3 read-back
+   verification** of the written objects (list the test bucket and GET a
+   sample — every read must succeed), per-node disk usage (observability),
   pass/fail verdict.

-Success requires **both** the heal API completion (the server's scan/repair
-verdict) and the outage node's disk reaching the target.
+Success is the server's own scan/repair verdict (heal finished, 0 failed)
+**plus** an end-to-end data read-back; per-node disk usage is logged as
+observability, not a pass gate (EC distributes different shards per node, so a
+fixed per-node GB target is not a meaningful invariant).

 ## Self-hosted runner prerequisites

@@ -66,7 +69,6 @@ Same repository secrets/variables as the pool expansion workflow:
 | `package_url`    | nightly | Direct `.deb` URL; empty = latest nightly |
 | `stop_node_gb`   | `15`    | Stop outage node at N GiB on survivors    |
 | `warp_stop_gb`   | `40`    | Stop warp at N GiB on survivors           |
-| `heal_target_gb` | `40`    | Outage node must reach N GiB after heal   |
 | `cleanup_before` | `true`  | Reset nodes before the test               |
 | `cleanup_after`  | `true`  | Reset nodes after the test                |

@@ -14,7 +14,8 @@
 #   4. Restart vm002 (the node that was offline while data was written)
 #   5. Start cluster heal (POST /rustfs/admin/v3/heal/ {"recursive":true})
 #   6. Monitor the heal task (POST /rustfs/admin/v3/heal/?clientToken=...)
-#      until a terminal success AND vm002's disk usage reaches HEAL_TARGET_GB
+#      until the server verdict is a terminal success (finished, 0 failed),
+#      then verify the data is readable back from the cluster
 #   7. Result analysis: heal stats, per-node disk usage, success verdict
 #
 # The script is driven from an admin host (e.g. a jumpbox or a GitHub
@@ -101,7 +102,6 @@ WARP_LOG_FILE="${RUSTFS_WARP_LOG_FILE:-}"
 # Disk-usage thresholds (per surviving node, GiB, via df -B1G | grep /data/rustfs)
 STOP_NODE_AT_GB="${RUSTFS_STOP_NODE_AT_GB:-15}"   # stop the outage node when surviving nodes reach this
 WARP_STOP_AT_GB="${RUSTFS_WARP_STOP_AT_GB:-40}"   # stop warp when surviving nodes reach this
-HEAL_TARGET_GB="${RUSTFS_HEAL_TARGET_GB:-40}"     # vm002 must reach this after heal to pass
 POLL_INTERVAL=15                # status polling interval (seconds)

 # Timeouts (seconds)
@@ -185,8 +185,9 @@ canonical_query() {
 # Issue an admin API request. Prints the response body on stdout and writes the
 # HTTP status (000 on transport failure) to ${ADMIN_API_CODE_FILE}.
 admin_api() {
-  # $1: method, $2: path, $3: query string, $4: optional JSON body
-  local method="$1" path="$2" query="$3" body="${4:-}"
+  # $1: method, $2: path, $3: query string, $4: optional JSON body,
+  # $5: "discard" to skip body capture (only the status code matters)
+  local method="$1" path="$2" query="$3" body="${4:-}" discard="${5:-}"
  local amz_date date_stamp host_port
  local canonical_headers signed_headers canonical_request string_to_sign
  local scope k_date k_region k_service k_signing signature auth
@@ -236,14 +237,18 @@ $(sha256_hex "${canonical_request}")"
  if [ -n "${body}" ]; then
    curl_body=(-d "${body}" -H "Content-Type: application/json")
  fi
-  code="$(curl -sS --max-time "${API_REQUEST_TIMEOUT}" -o "${tmp}" -w '%{http_code}' \
+  local out_file="${tmp}"
+  [ "${discard}" = "discard" ] && out_file="/dev/null"
+  code="$(curl -sS --max-time "${API_REQUEST_TIMEOUT}" -o "${out_file}" -w '%{http_code}' \
    -H "Host: ${host_port}" \
    -H "x-amz-content-sha256: UNSIGNED-PAYLOAD" \
    -H "x-amz-date: ${amz_date}" \
    -H "Authorization: ${auth}" \
    "${curl_body[@]}" -X "${method}" "${url}")" || code="000"
  printf '%s' "${code}" > "${ADMIN_API_CODE_FILE}"
-  cat "${tmp}"
+  if [ "${discard}" != "discard" ]; then
+    cat "${tmp}"
+  fi
  rm -f "${tmp}"
 }

@@ -531,6 +536,42 @@ heal_progress_field() {
    | if $p == null then "null" else (($p[$c] // $p[$s] // null) | if . == null then "null" else tostring end) end'
 }

+# Sample data verification after heal: list the test bucket and GET a sample of
+# objects. Every GET must return 200 — this is the end-to-end proof that the
+# cluster can still reconstruct the data after repair.
+verify_data_readable() {
+  local list body code keys key count checked ok
+  if [ "${DRY_RUN}" -eq 1 ]; then
+    log "DRY-RUN: S3 read-back verification of ${WARP_BUCKET}"
+    return 0
+  fi
+  list="$(admin_api GET "/${WARP_BUCKET}" "list-type=2&max-keys=1000")"
+  code="$(admin_api_code)"
+  if [ "${code}" != "200" ]; then
+    printf '\033[1;31m[ERROR]\033[0m bucket list failed (HTTP %s): %s\n' "${code}" "${list}" >&2
+    return 1
+  fi
+  # sed -n '1,20p' reads the whole stream (unlike head, which closes the pipe
+  # early and SIGPIPEs grep/sed under pipefail).
+  keys="$(printf '%s' "${list}" | grep -oE '<Key>[^<]+</Key>' | sed 's#</\?Key>##g' | sed -n '1,20p')"
+  count="$(printf '%s\n' "${keys}" | sed '/^$/d' | wc -l | tr -d ' ')"
+  log "data verification: ${count} object(s) sampled from the bucket; reading each (status-code check)..."
+  checked=0; ok=0
+  while IFS= read -r key; do
+    [ -z "${key}" ] && continue
+    admin_api GET "/${WARP_BUCKET}/${key}" "" "" discard
+    code="$(admin_api_code)"
+    checked=$((checked + 1))
+    if [ "${code}" = "200" ]; then
+      ok=$((ok + 1))
+    else
+      printf '\033[1;31m[ERROR]\033[0m GET %s failed (HTTP %s)\n' "${key}" "${code}" >&2
+    fi
+  done <<<"${keys}"
+  log "data verification: ${ok}/${checked} objects read successfully"
+  [ "${checked}" -gt 0 ] && [ "${ok}" -eq "${checked}" ]
+}
+
 # Verify the expected number of pools via the admin API (JSON + jq assertions)
 verify_pools() {
  local expected="$1"
@@ -847,7 +888,7 @@ step5_start_heal() {
 }

 step6_monitor_heal() {
-  log "step 6: monitor heal task until done AND ${NODES[${OUTAGE_NODE_INDEX}]} reaches ${HEAL_TARGET_GB}GB"
+  log "step 6: monitor heal task until the server verdict is done"
  if [ "${DRY_RUN}" -eq 1 ]; then
    log "DRY-RUN: waiting for heal to complete"
    return 0
@@ -856,7 +897,9 @@ step6_monitor_heal() {
    die "no heal client token (run step 5 first, or pass --heal-token)"
  fi
  local waited=0 body code summary failed healed scanned pct prog_present
-  local failed_n healed_n scanned_n pct_n vm002_used warned501=0
+  local failed_n healed_n scanned_n pct_n vm002_used warned501=0 pre_outage_used
+  pre_outage_used="$(node_used_gb "${NODES[${OUTAGE_NODE_INDEX}]}")"
+  log "outage node usage before heal: ${pre_outage_used}GB"
  while [ "${waited}" -lt "${HEAL_TIMEOUT}" ]; do
    body="$(admin_api POST /rustfs/admin/v3/heal/ "clientToken=${HEAL_CLIENT_TOKEN}" "")"
    code="$(admin_api_code)"
@@ -882,7 +925,7 @@ step6_monitor_heal() {
    scanned_n="${scanned}"; [ "${scanned_n}" = "null" ] && scanned_n=0
    pct_n="${pct}"; [ "${pct_n}" = "null" ] && pct_n=0
    vm002_used="$(node_used_gb "${NODES[${OUTAGE_NODE_INDEX}]}")"
-    log "heal: summary=${summary} scanned=${scanned_n} healed=${healed_n} failed=${failed_n} pct=${pct_n} ${NODES[${OUTAGE_NODE_INDEX}]}_used=${vm002_used}GB (target ${HEAL_TARGET_GB}GB)"
+    log "heal: summary=${summary} scanned=${scanned_n} healed=${healed_n} failed=${failed_n} pct=${pct_n} ${NODES[${OUTAGE_NODE_INDEX}]}_used=${vm002_used}GB"

    if [ "${prog_present}" = "false" ] && [ "${summary}" = "running" ]; then
      warn "heal progress is null while the task is running (server-side reporting gap; see rustfs/backlog#2035)"
@@ -897,9 +940,11 @@ step6_monitor_heal() {

    # Success only for a real terminal summary; "running"/"notFound"/"" mean
    # the task is still going (or lives on another node) — keep polling.
-    if printf '%s' "${summary}" | grep -qiE '^(finished|completed|success|done)$' \
-      && [ "${vm002_used}" -ge "${HEAL_TARGET_GB}" ]; then
-      log "heal done: summary=${summary} failed=0 ${NODES[${OUTAGE_NODE_INDEX}]}_used=${vm002_used}GB >= ${HEAL_TARGET_GB}GB"
+    if printf '%s' "${summary}" | grep -qiE '^(finished|completed|success|done)$'; then
+      log "heal done: summary=${summary} failed=0 (server verdict)"
+      if [ "${vm002_used}" -le "${pre_outage_used}" ]; then
+        warn "heal finished but ${NODES[${OUTAGE_NODE_INDEX}]} usage did not grow (${pre_outage_used}GB -> ${vm002_used}GB); the repair may not have landed on its disks"
+      fi
      final_status_file="$(mktemp "${TMPDIR:-/tmp}/rustfs-heal-final-status.XXXXXX.json" 2>/dev/null \
        || printf '%s' "${TMPDIR:-/tmp}/rustfs-heal-final-status.$$.json")"
      printf '%s\n' "${body}" > "${final_status_file}" 2>/dev/null \
@@ -907,6 +952,7 @@ step6_monitor_heal() {
        || warn "could not save final heal status to ${final_status_file}"
      return 0
    fi
+
    sleep "${POLL_INTERVAL}"
    waited=$((waited + POLL_INTERVAL))
  done
@@ -942,9 +988,6 @@ step7_analyze_results() {
  printf '%s\n' "--- heal status ---"
  printf '  summary=%s scanned=%s healed=%s failed=%s progress=%s%%\n' \
    "${summary}" "${scanned_n}" "${healed_n}" "${failed_n}" "${pct_n}"
-  if [ "${prog_present}" = "false" ]; then
-    warn "heal progress was absent (null) in the final task response — see rustfs/backlog#2035"
-  fi
  printf '%s\n' "--- per-node disk usage (GiB) ---"
  for i in "${!NODES[@]}"; do
    used="$(node_used_gb "${NODES[$i]}")"
@@ -953,13 +996,17 @@ step7_analyze_results() {

  local outage_used
  outage_used="$(node_used_gb "${NODES[${OUTAGE_NODE_INDEX}]}")"
+
+  if ! verify_data_readable; then
+    die "heal test FAILED: data read-back verification failed (see errors above)"
+  fi
+
  if printf '%s' "${summary}" | grep -qiE '^(finished|completed|success|done)$' \
-    && [ "${failed_n}" -eq 0 ] \
-    && [ "${outage_used}" -ge "${HEAL_TARGET_GB}" ]; then
-    log "heal test PASSED: cluster heal complete, 0 failed, ${NODES[${OUTAGE_NODE_INDEX}]} reached ${outage_used}GB"
+    && [ "${failed_n}" -eq 0 ]; then
+    log "heal test PASSED: cluster heal complete, 0 failed, data read-back OK, ${NODES[${OUTAGE_NODE_INDEX}]}_used=${outage_used}GB"
    return 0
  fi
-  die "heal test FAILED: summary=${summary} failed=${failed_n} ${NODES[${OUTAGE_NODE_INDEX}]}_used=${outage_used}GB (target ${HEAL_TARGET_GB}GB)"
+  die "heal test FAILED: summary=${summary} failed=${failed_n} ${NODES[${OUTAGE_NODE_INDEX}]}_used=${outage_used}GB"
 }
 # ==================== CLI parsing ====================

@@ -983,7 +1030,6 @@ Options:
  --rc-endpoint URL          Deprecated alias for --endpoint
  --stop-node-gb N           Stop the outage node when surviving nodes reach N GiB (default 15)
  --warp-stop-gb N           Stop warp when surviving nodes reach N GiB (default 40)
-  --heal-target-gb N         Outage node must reach N GiB after heal (default 40)
  --heal-token TOKEN         clientToken of a heal started earlier (for steps 6/7 reruns)
  --warp-timeout N           Write phase timeout in seconds (default 3600)
  --heal-timeout N           Heal wait timeout in seconds (default 86400)
@@ -1056,7 +1102,6 @@ main() {
      --rc-endpoint) API_ENDPOINT="$1"; shift ;;
      --stop-node-gb) STOP_NODE_AT_GB="$1"; shift ;;
      --warp-stop-gb) WARP_STOP_AT_GB="$1"; shift ;;
-      --heal-target-gb) HEAL_TARGET_GB="$1"; shift ;;
      --heal-token) HEAL_CLIENT_TOKEN="$1"; shift ;;
      --warp-timeout) WARP_TIMEOUT="$1"; shift ;;
      --heal-timeout) HEAL_TIMEOUT="$1"; shift ;;
@@ -0,0 +1,543 @@
+#!/usr/bin/env bash
+#
+# rustfs-performance-test.sh
+# RustFS 4x4 集群性能压测全流程脚本
+#
+# Based on the Obsidian note "RustFS 性能测试". Full workflow:
+#   1. Cleanup: stop & purge rustfs, remove data dirs on all nodes
+#   2. Download the RustFS package on all nodes
+#   3. Install RustFS on all nodes (dpkg -i), recreate volume dirs
+#   4. Write /etc/default/rustfs (4-node x 4-drive MNMD), start all nodes
+#      in parallel and verify the service is Running
+#   5. Run the benchmark (warp GET/PUT/MIXED via rustfs-performance-testing.sh)
+#   6. Analyze results (summary.tsv / summary.md)
+#   7. Final cleanup: stop & purge rustfs, remove data dirs
+#
+# The script is driven from an admin host (e.g. a jumpbox) and operates on
+# the target nodes over SSH, mirroring scripts/test/rustfs_*_test.sh.
+#
+# Usage:
+#   ./rustfs-performance-test.sh --all                  # run all steps 1-7
+#   ./rustfs-performance-test.sh --step 5               # run a single step
+#   ./rustfs-performance-test.sh --steps 2,3,4          # run selected steps
+#   ./rustfs-performance-test.sh --all --dry-run        # preview only
+#   ./rustfs-performance-test.sh --all -y --package-url <deb URL>
+#
+# Notes:
+#   - SSH user defaults to azureuser (passwordless sudo on the nodes);
+#     pass --ssh-user root if your nodes accept root login.
+#   - The benchmark runner defaults to
+#     ~/Documents/Obsidian Vault/rustfs-performance-testing.sh; override with
+#     --bench-script / RUSTFS_BENCH_SCRIPT. warp must be installed on the
+#     admin host.
+#   - Steps 1 and 7 destroy the RustFS install and all data (confirmed).
+#
+set -Eeuo pipefail
+
+# ==================== Configuration (adjust to your environment) ====================
+
+# Target nodes (4x4: 4 nodes x 4 drives each)
+if [ -n "${RUSTFS_NODES:-}" ]; then
+  read -r -a NODES <<<"${RUSTFS_NODES}"
+else
+  NODES=(vm000 vm001 vm002 vm003)
+fi
+
+SSH_USER="${RUSTFS_SSH_USER:-azureuser}"
+SSH_PORT="${RUSTFS_SSH_PORT:-22}"
+SSH_OPTS=(-o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new -p "${SSH_PORT}")
+
+# Package: GitHub release tag, e.g. "1.0.0-rc.3". PACKAGE_URL is derived from
+# RUSTFS_VERSION unless --package-url / RUSTFS_PACKAGE_URL is given.
+RUSTFS_VERSION="${RUSTFS_VERSION:-1.0.0-rc.3}"
+PACKAGE_URL="${RUSTFS_PACKAGE_URL:-}"
+ARCH="${RUSTFS_ARCH:-amd64}"
+PACKAGES_DIR="/home/rustfs/packages"
+PACKAGE_FILE="rustfs.deb"
+PACKAGE_SHA256="${RUSTFS_PACKAGE_SHA256:-}"
+
+# 4x4 topology: 4 nodes x 4 drives each, same expression on every node
+DRIVES_PER_NODE="${RUSTFS_DRIVES_PER_NODE:-4}"
+VOLUMES="http://rustfs-node{1...4}:9000/data/rustfs{1...4}/mnmd"
+
+# RustFS service configuration (written to /etc/default/rustfs)
+RUSTFS_CONFIG_FILE="/etc/default/rustfs"
+RUSTFS_SERVICE="rustfs"
+RUSTFS_PACKAGE_NAME="rustfs"
+RUSTFS_USER="rustfs"
+ACCESS_KEY="${RUSTFS_ACCESS_KEY:-rustfs@test}"
+SECRET_KEY="${RUSTFS_SECRET_KEY:-rustfs@test}"
+RUSTFS_ADDRESS=":9000"
+RUSTFS_CONSOLE_ADDRESS=":9001"
+RUSTFS_CONSOLE_ENABLE=true
+RUSTFS_OBS_LOGGER_LEVEL=error
+RUSTFS_OBS_LOG_DIRECTORY="/var/log/rustfs/"
+
+# Benchmark runner (step 5/6): prefer the default Obsidian location, fall back
+# to a rustfs-performance-testing.sh next to this script (e.g. in the repo or
+# on a jumpbox).
+_DEFAULT_BENCH="${HOME}/Documents/Obsidian Vault/rustfs-performance-testing.sh"
+_SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+if [ -x "${_DEFAULT_BENCH}" ]; then
+  _BENCH_RESOLVED="${_DEFAULT_BENCH}"
+elif [ -x "${_SCRIPT_DIR}/rustfs-performance-testing.sh" ]; then
+  _BENCH_RESOLVED="${_SCRIPT_DIR}/rustfs-performance-testing.sh"
+elif [ -x "${_SCRIPT_DIR}/rustfs_performance_testing.sh" ]; then
+  _BENCH_RESOLVED="${_SCRIPT_DIR}/rustfs_performance_testing.sh"
+else
+  _BENCH_RESOLVED="${_DEFAULT_BENCH}"
+fi
+BENCH_SCRIPT="${RUSTFS_BENCH_SCRIPT:-${_BENCH_RESOLVED}}"
+RESULT_DIR="${RUSTFS_RESULT_DIR:-$(pwd)/warp-bench-results-$(date +%Y%m%d-%H%M%S)}"
+WARP_HOST="${RUSTFS_WARP_HOST:-rustfs-node1:9000,rustfs-node2:9000,rustfs-node3:9000,rustfs-node4:9000}"
+WARP_BUCKET="${RUSTFS_WARP_BUCKET:-warp-benchmark-bucket}"
+WARP_CONCURRENCY="${RUSTFS_WARP_CONCURRENCY:-64}"
+WARP_DURATION="${RUSTFS_WARP_DURATION:-5m}"
+WARP_GET_OBJECTS="${RUSTFS_WARP_GET_OBJECTS:-2500}"
+WARP_SLEEP="${RUSTFS_WARP_SLEEP:-60}"
+# Manual method/size selection (passed through to the benchmark runner; empty = full run)
+WARP_METHODS="${RUSTFS_WARP_METHODS:-}"
+WARP_SIZES="${RUSTFS_WARP_SIZES:-}"
+
+# Timeouts (seconds)
+SERVICE_TIMEOUT="${RUSTFS_SERVICE_TIMEOUT:-300}"
+POLL_INTERVAL="${RUSTFS_POLL_INTERVAL:-10}"
+
+# ==================== Runtime options (set by CLI) ====================
+DRY_RUN=0
+ASSUME_YES=0
+SKIP_DOWNLOAD=0
+PREFLIGHT=0
+LOG_FILE=""
+SELECTED_STEPS=()
+
+# ==================== Helpers ====================
+
+log()  { printf '\033[1;36m[INFO]\033[0m %s\n' "$*"; }
+warn() { printf '\033[1;33m[WARN]\033[0m %s\n' "$*"; }
+die()  { printf '\033[1;31m[ERROR]\033[0m %s\n' "$*" >&2; exit 1; }
+
+confirm() {
+  if [ "${ASSUME_YES}" -eq 1 ] || [ "${DRY_RUN}" -eq 1 ]; then return 0; fi
+  printf '\033[1;33m[CONFIRM]\033[0m %s (y/N) ' "$1"
+  read -r answer
+  case "${answer}" in
+    y|Y|yes|YES) return 0 ;;
+    *) die "cancelled" ;;
+  esac
+}
+
+need_cmd() {
+  [ "${DRY_RUN}" -eq 1 ] && return 0
+  command -v "$1" >/dev/null 2>&1 || die "missing command: $1 ($2); install it first"
+}
+
+# Run a remote script on a single node (script is read from stdin)
+run_remote() {
+  local node="$1" script
+  script="$(cat)"
+  if [ "${DRY_RUN}" -eq 1 ]; then
+    log "DRY-RUN: ssh ${SSH_USER}@${node} <<'REMOTE'"
+    printf '%s\n' "${script}" | sed 's/^/    | /'
+    log "DRY-RUN: ----"
+    return 0
+  fi
+  log "==> ${node}: executing remote script"
+  ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" 'bash -s' <<<"${script}"
+}
+
+# Run the same remote script on all nodes in parallel (script from stdin)
+run_remote_all() {
+  local script pids=() i=0 fail=0
+  script="$(cat)"
+  for node in "${NODES[@]}"; do
+    if [ "${DRY_RUN}" -eq 1 ]; then
+      log "DRY-RUN: ssh ${SSH_USER}@${node} <<'REMOTE'"
+      printf '%s\n' "${script}" | sed 's/^/    | /'
+      log "DRY-RUN: ----"
+    else
+      log "==> ${node}: executing remote script"
+      ( ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" 'bash -s' <<<"${script}" ) &
+      pids[$i]=$!
+      i=$((i+1))
+    fi
+  done
+  if [ "${#pids[@]}" -gt 0 ]; then
+    for pid in "${pids[@]}"; do
+      wait "${pid}" || fail=1
+    done
+  fi
+  [ "${fail}" -eq 0 ] || die "one or more remote executions failed"
+}
+
+rustfs_config_body() {
+  cat <<EOF
+RUSTFS_ACCESS_KEY=${ACCESS_KEY}
+RUSTFS_SECRET_KEY=${SECRET_KEY}
+RUSTFS_VOLUMES="${VOLUMES}"
+RUSTFS_ADDRESS="${RUSTFS_ADDRESS}"
+RUSTFS_CONSOLE_ADDRESS="${RUSTFS_CONSOLE_ADDRESS}"
+RUSTFS_CONSOLE_ENABLE=${RUSTFS_CONSOLE_ENABLE}
+RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL}
+RUSTFS_OBS_LOG_DIRECTORY="${RUSTFS_OBS_LOG_DIRECTORY}"
+EOF
+}
+
+write_rustfs_config() {
+  local node="$1" body
+  body="$(rustfs_config_body)"
+  log "${node}: writing config ${RUSTFS_CONFIG_FILE}"
+  {
+    printf 'set -euo pipefail\n'
+    printf 'SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"\n'
+    printf '%s tee %s >/dev/null <<RUSTFS_EOF\n' '${SUDO}' "${RUSTFS_CONFIG_FILE}"
+    printf '%s' "${body}"
+    printf '\nRUSTFS_EOF\n'
+    printf '${SUDO} systemctl daemon-reload\n'
+  } | run_remote "${node}"
+}
+
+service_action() {
+  local action="$1" node="$2"
+  log "${node}: systemctl ${action} ${RUSTFS_SERVICE}"
+  [ "${DRY_RUN}" -eq 1 ] && return 0
+  ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \
+    "if [ \"\$(id -u)\" -ne 0 ]; then sudo -n systemctl ${action} ${RUSTFS_SERVICE}; else systemctl ${action} ${RUSTFS_SERVICE}; fi" \
+    || die "${node}: systemctl ${action} failed"
+}
+
+wait_service_active() {
+  local node="$1" elapsed=0
+  log "${node}: waiting for ${RUSTFS_SERVICE} to become active"
+  [ "${DRY_RUN}" -eq 1 ] && { log "${node}: (dry-run) skip wait"; return 0; }
+  while [ "${elapsed}" -lt "${SERVICE_TIMEOUT}" ]; do
+    if ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \
+        "systemctl is-active ${RUSTFS_SERVICE} 2>/dev/null" | grep -q active; then
+      log "${node}: service active"
+      return 0
+    fi
+    sleep "${POLL_INTERVAL}"
+    elapsed=$((elapsed + POLL_INTERVAL))
+  done
+  die "${node}: ${RUSTFS_SERVICE} did not become active within ${SERVICE_TIMEOUT}s"
+}
+
+verify_service_running() {
+  local node="$1"
+  log "${node}: checking service status"
+  if [ "${DRY_RUN}" -eq 0 ]; then
+    ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \
+      "systemctl status ${RUSTFS_SERVICE} --no-pager | head -n 12" || true
+  fi
+}
+
+build_package_url() {
+  local asset
+  asset="rustfs_$(printf '%s' "${RUSTFS_VERSION}" | tr '-' '.')_${ARCH}.deb"
+  printf 'https://github.com/rustfs/rustfs/releases/download/%s/%s' "${RUSTFS_VERSION}" "${asset}"
+}
+
+resolve_package_url() {
+  if [ -n "${PACKAGE_URL}" ]; then printf '%s' "${PACKAGE_URL}"; else build_package_url; fi
+}
+
+preflight() {
+  log "preflight checks"
+  need_cmd ssh "openssh client"
+  need_cmd curl "http client"
+  need_cmd warp "warp benchmark tool (for step 5)"
+  if [ ! -x "${BENCH_SCRIPT}" ]; then
+    die "benchmark script not found or not executable: ${BENCH_SCRIPT}"
+  fi
+  if [ "${DRY_RUN}" -eq 0 ]; then
+    for node in "${NODES[@]}"; do
+      ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" 'echo ok' >/dev/null \
+        || die "cannot ssh to ${node}"
+    done
+    log "all nodes reachable: ${NODES[*]}"
+  fi
+  log "preflight OK"
+}
+
+# ==================== Steps ====================
+
+step1_cleanup() {
+  log "step 1: cleanup environment on all nodes (stop & purge rustfs, remove data dirs)"
+  confirm "This DESTROYS the RustFS install and ALL data on ${NODES[*]} (irreversible). Continue?"
+  local script
+  script="$(cat <<EOF
+set -euo pipefail
+SUDO=""; [ "\$(id -u)" -ne 0 ] && SUDO="sudo -n"
+\${SUDO} systemctl stop rustfs 2>/dev/null || true
+if \${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
+  \${SUDO} dpkg -P rustfs
+  echo "purged rustfs"
+else
+  echo "rustfs not installed, skip purge"
+fi
+for i in \$(seq 1 ${DRIVES_PER_NODE}); do
+  \${SUDO} rm -rf /data/rustfs\${i}/mnmd
+  \${SUDO} mkdir -p /data/rustfs\${i}/mnmd
+  \${SUDO} chown -R rustfs:rustfs /data/rustfs\${i}/mnmd
+done
+echo "cleanup done on \$(hostname)"
+EOF
+)"
+  printf '%s\n' "${script}" | run_remote_all
+  log "step 1 complete"
+}
+
+step2_download() {
+  log "step 2: download the package on all nodes"
+  local url script
+  url="$(resolve_package_url)"
+  script="$(cat <<EOF
+set -euo pipefail
+SUDO=""; [ "\$(id -u)" -ne 0 ] && SUDO="sudo -n"
+if [ -f "${PACKAGES_DIR}/${PACKAGE_FILE}" ] && [ "${SKIP_DOWNLOAD}" -eq 1 ]; then
+  echo "already exists: ${PACKAGES_DIR}/${PACKAGE_FILE}, skipping download"
+else
+  echo "downloading ${url} ..."
+  curl -fSL --retry 3 -o "/tmp/${PACKAGE_FILE}" "${url}"
+  \${SUDO} mkdir -p "${PACKAGES_DIR}"
+  \${SUDO} install -m 0644 "/tmp/${PACKAGE_FILE}" "${PACKAGES_DIR}/${PACKAGE_FILE}"
+  \${SUDO} rm -f "/tmp/${PACKAGE_FILE}"
+fi
+if [ -n "${PACKAGE_SHA256}" ]; then
+  echo "${PACKAGE_SHA256}  ${PACKAGES_DIR}/${PACKAGE_FILE}" | sha256sum -c - || { echo "checksum verification failed"; exit 1; }
+fi
+ls -lh "${PACKAGES_DIR}/${PACKAGE_FILE}"
+EOF
+)"
+  printf '%s\n' "${script}" | run_remote_all
+  log "step 2 complete"
+}
+
+step3_install() {
+  log "step 3: install the RustFS service on all nodes"
+  confirm "About to run dpkg -i ${PACKAGE_FILE} on all nodes. Continue?"
+  local script
+  script="$(cat <<'EOF'
+set -euo pipefail
+SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
+${SUDO} dpkg -i /home/rustfs/packages/rustfs.deb
+${SUDO} systemctl daemon-reload
+echo "--- installed package ---"
+dpkg -l rustfs | tail -n 1
+EOF
+)"
+  printf '%s\n' "${script}" | run_remote_all
+  log "step 3 complete"
+}
+
+step4_configure_start() {
+  log "step 4: write config, start and verify the service on all nodes"
+  local node
+  for node in "${NODES[@]}"; do
+    write_rustfs_config "${node}"
+  done
+  for node in "${NODES[@]}"; do
+    service_action start "${node}" &
+  done
+  wait
+  for node in "${NODES[@]}"; do
+    wait_service_active "${node}"
+    verify_service_running "${node}"
+  done
+  log "step 4 complete"
+}
+
+step5_benchmark() {
+  log "step 5: run the benchmark (${BENCH_SCRIPT})"
+  need_cmd warp "warp benchmark tool"
+  confirm "About to run the full GET/PUT/MIXED benchmark (~6-10 hours). Continue?"
+  if [ "${DRY_RUN}" -eq 1 ]; then
+    log "DRY-RUN: WARP_HOST=${WARP_HOST} WARP_RESULT_DIR=${RESULT_DIR} bash ${BENCH_SCRIPT}"
+    return 0
+  fi
+  WARP_HOST="${WARP_HOST}" \
+  WARP_ACCESS_KEY="${ACCESS_KEY}" \
+  WARP_SECRET_KEY="${SECRET_KEY}" \
+  WARP_BUCKET="${WARP_BUCKET}" \
+  WARP_CONCURRENCY="${WARP_CONCURRENCY}" \
+  WARP_DURATION="${WARP_DURATION}" \
+  WARP_GET_OBJECTS="${WARP_GET_OBJECTS}" \
+  WARP_SLEEP_BETWEEN_ROUNDS="${WARP_SLEEP}" \
+  WARP_METHODS="${WARP_METHODS}" \
+  WARP_SIZES="${WARP_SIZES}" \
+  WARP_RESULT_DIR="${RESULT_DIR}" \
+  bash "${BENCH_SCRIPT}"
+  log "step 5 complete (results in ${RESULT_DIR})"
+}
+
+step6_analyze() {
+  log "step 6: analyze results from ${RESULT_DIR}"
+  if [ "${DRY_RUN}" -eq 1 ]; then
+    log "DRY-RUN: bash ${BENCH_SCRIPT} --parse-only ${RESULT_DIR}"
+    return 0
+  fi
+  if [ ! -d "${RESULT_DIR}" ]; then
+    die "result directory not found: ${RESULT_DIR}"
+  fi
+  if [ -f "${RESULT_DIR}/summary.md" ]; then
+    log "summary already generated: ${RESULT_DIR}/summary.md"
+  else
+    log "generating summary with --parse-only"
+    bash "${BENCH_SCRIPT}" --parse-only "${RESULT_DIR}"
+  fi
+  log "----- summary.md -----"
+  cat "${RESULT_DIR}/summary.md"
+  log "step 6 complete"
+}
+
+step7_cleanup() {
+  log "step 7: final cleanup on all nodes (stop & purge rustfs, remove data dirs)"
+  confirm "This DESTROYS the RustFS install and ALL data on ${NODES[*]} (irreversible). Continue?"
+  local script
+  script="$(cat <<EOF
+set -euo pipefail
+SUDO=""; [ "\$(id -u)" -ne 0 ] && SUDO="sudo -n"
+\${SUDO} systemctl stop rustfs 2>/dev/null || true
+if \${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
+  \${SUDO} dpkg -P rustfs
+  echo "purged rustfs"
+fi
+for i in \$(seq 1 ${DRIVES_PER_NODE}); do
+  \${SUDO} rm -rf /data/rustfs\${i}/mnmd
+done
+echo "cleanup done on \$(hostname)"
+EOF
+)"
+  printf '%s\n' "${script}" | run_remote_all
+  log "step 7 complete"
+}
+
+# ==================== CLI ====================
+
+usage() {
+  cat <<'USAGE'
+Usage: ./rustfs-performance-test.sh [options]
+
+Steps:
+  1  cleanup environment (purge rustfs, remove data dirs)   [destructive]
+  2  download the RustFS package on all nodes
+  3  install RustFS (dpkg -i)
+  4  write config, start service, verify Running
+  5  run benchmark (warp GET/PUT/MIXED)
+  6  analyze results (summary.tsv / summary.md)
+  7  final cleanup (purge rustfs, remove data dirs)         [destructive]
+
+Options:
+  --all                     Run all steps 1-7
+  --step N                  Run a single step
+  --steps 1,3,5-7           Run selected steps
+  --version VERSION         GitHub release tag (default 1.0.0-rc.3)
+  --package-url URL         Direct deb URL (overrides --version)
+  --sha256 HASH             Verify package checksum
+  --skip-download           Keep an existing package file
+  --bench-script PATH       Benchmark runner (default: Obsidian Vault rustfs-performance-testing.sh)
+  --result-dir DIR          Benchmark result directory
+  --warp-duration DUR       warp duration per round (default 5m)
+  --warp-concurrency N      warp concurrency (default 64)
+  --ssh-user USER           SSH user (default azureuser)
+  --ssh-port PORT           SSH port (default 22)
+  --preflight               Check environment and exit
+  --log-file FILE           Append all output to FILE
+  --dry-run                 Preview commands without executing them
+  -y, --yes                 Skip all confirmation prompts
+  -h, --help                Show this help
+
+Examples:
+  ./rustfs-performance-test.sh --all
+  ./rustfs-performance-test.sh --all --dry-run
+  ./rustfs-performance-test.sh --all -y --package-url https://dl.rustfs.com/...deb
+  ./rustfs-performance-test.sh --step 5
+USAGE
+}
+
+expand_steps() {
+  local spec="$1" part start end i
+  IFS=',' read -ra parts <<<"${spec}"
+  for part in "${parts[@]}"; do
+    if [[ "${part}" =~ ^([0-9]+)-([0-9]+)$ ]]; then
+      start="${BASH_REMATCH[1]}"; end="${BASH_REMATCH[2]}"
+      for ((i=start; i<=end; i++)); do SELECTED_STEPS+=("${i}"); done
+    elif [[ "${part}" =~ ^[0-9]+$ ]]; then
+      SELECTED_STEPS+=("${part}")
+    else
+      die "cannot parse step spec: ${part}"
+    fi
+  done
+}
+
+run_steps() {
+  local step
+  for step in "${SELECTED_STEPS[@]}"; do
+    case "${step}" in
+      1) step1_cleanup ;;
+      2) step2_download ;;
+      3) step3_install ;;
+      4) step4_configure_start ;;
+      5) step5_benchmark ;;
+      6) step6_analyze ;;
+      7) step7_cleanup ;;
+      *) die "unknown step: ${step}" ;;
+    esac
+    log "step ${step} completed"
+  done
+}
+
+main() {
+  [ "$#" -eq 0 ] && { usage; exit 0; }
+  local opt all=0
+  while [ "$#" -gt 0 ]; do
+    opt="$1"; shift
+    case "${opt}" in
+      --all) all=1 ;;
+      --step) SELECTED_STEPS+=("$1"); shift ;;
+      --steps) expand_steps "$1"; shift ;;
+      --version) RUSTFS_VERSION="$1"; shift ;;
+      --package-url) PACKAGE_URL="$1"; shift ;;
+      --sha256) PACKAGE_SHA256="$1"; shift ;;
+      --skip-download) SKIP_DOWNLOAD=1 ;;
+      --bench-script) BENCH_SCRIPT="$1"; shift ;;
+      --result-dir) RESULT_DIR="$1"; shift ;;
+      --warp-duration) WARP_DURATION="$1"; shift ;;
+      --warp-concurrency) WARP_CONCURRENCY="$1"; shift ;;
+      --ssh-user) SSH_USER="$1"; shift ;;
+      --ssh-port) SSH_PORT="$1"; shift ;;
+      --preflight) PREFLIGHT=1 ;;
+      --log-file) LOG_FILE="$1"; shift ;;
+      --dry-run) DRY_RUN=1 ;;
+      -y|--yes) ASSUME_YES=1 ;;
+      -h|--help) usage; exit 0 ;;
+      *) die "unknown option: ${opt} (see --help)" ;;
+    esac
+  done
+  if [ -n "${LOG_FILE}" ]; then
+    mkdir -p "$(dirname "${LOG_FILE}")"
+    exec > >(tee -a "${LOG_FILE}") 2>&1
+  fi
+  if [ "${all}" -eq 1 ]; then
+    SELECTED_STEPS=(1 2 3 4 5 6 7)
+  fi
+  if [ "${PREFLIGHT}" -eq 1 ]; then
+    preflight
+    if [ "${#SELECTED_STEPS[@]}" -eq 0 ]; then
+      log "preflight only; done"
+      exit 0
+    fi
+  fi
+  [ "${#SELECTED_STEPS[@]}" -gt 0 ] || die "no steps selected (--all / --step / --steps)"
+  log "nodes: ${NODES[*]}  ssh user: ${SSH_USER}  version: ${RUSTFS_VERSION}"
+  log "package: $(resolve_package_url)"
+  log "result dir: ${RESULT_DIR}"
+  [ "${DRY_RUN}" -eq 1 ] && warn "DRY-RUN mode: only printing the commands that would run"
+  run_steps
+  log "all done"
+}
+
+# Allow sourcing the file for unit tests without running main.
+if [ "${RUSTFS_PERF_SCRIPT_SOURCE_ONLY:-0}" != "1" ]; then
+  main "$@"
+fi
@@ -0,0 +1,239 @@
+#!/usr/bin/env bash
+#
+# rustfs-performance-testing.sh
+# RustFS 对象存储压测脚本（固定版）：测试方法 + 执行 + 结果解析
+#
+# 测试方法
+#   1) 方法：GET / PUT / MIXED（warp 默认混合负载 45% GET + 55% PUT）
+#   2) 对象尺寸：1KiB 4KiB 16KiB 128KiB 1MiB 4MiB 8MiB 16MiB 32MiB 64MiB
+#   3) 并发：64；单轮时长：5m；轮间 sleep：60s；GET 对象数：2500
+#   4) 顺序：GET 全部尺寸 -> PUT 全部尺寸 -> MIXED 全部尺寸
+#   5) 结果解析：每轮结束后自动解析 warp 输出，写入
+#      summary.tsv（机器可读）与 summary.md（Markdown 汇总表）
+#
+# 依赖：warp >= v1.6（MinIO warp），bash，awk/sed/grep
+# 说明：warp v1.6.1 的 put 不支持 --objects，脚本已自动处理（仅 get/mixed 传该参数）
+#
+# 环境变量覆盖（不传时使用固定默认值）：
+#   WARP_HOST WARP_ACCESS_KEY WARP_SECRET_KEY WARP_BUCKET
+#   WARP_CONCURRENCY WARP_DURATION WARP_GET_OBJECTS WARP_SLEEP_BETWEEN_ROUNDS
+#   WARP_RESULT_DIR
+#   WARP_METHODS WARP_SIZES   # 手动指定方法/尺寸（逗号或空格分隔），不传则全量
+
+set -u -o pipefail
+
+HOST="${WARP_HOST:-rustfs-node1:9000,rustfs-node2:9000,rustfs-node3:9000,rustfs-node4:9000}"
+ACCESS_KEY="${WARP_ACCESS_KEY:-rustfs@test}"
+SECRET_KEY="${WARP_SECRET_KEY:-rustfs@test}"
+BUCKET="${WARP_BUCKET:-warp-benchmark-bucket}"
+CONCURRENCY="${WARP_CONCURRENCY:-64}"
+DURATION="${WARP_DURATION:-5m}"
+GET_OBJECTS="${WARP_GET_OBJECTS:-2500}"
+SLEEP_BETWEEN_ROUNDS="${WARP_SLEEP_BETWEEN_ROUNDS:-60}"
+RESULT_DIR="${WARP_RESULT_DIR:-$(pwd)/warp-bench-results-$(date +%Y%m%d-%H%M%S)}"
+
+if [ -n "${WARP_SIZES:-}" ] && [ "${WARP_SIZES}" != "all" ] && [ "${WARP_SIZES}" != "ALL" ]; then
+  read -r -a SIZES <<<"${WARP_SIZES//,/ }"
+else
+  SIZES=(1KiB 4KiB 16KiB 128KiB 1MiB 4MiB 8MiB 16MiB 32MiB 64MiB)
+fi
+if [ -n "${WARP_METHODS:-}" ] && [ "${WARP_METHODS}" != "all" ] && [ "${WARP_METHODS}" != "ALL" ]; then
+  read -r -a METHODS <<<"${WARP_METHODS//,/ }"
+else
+  METHODS=(get put mixed)
+fi
+TOTAL_ROUNDS=$(( ${#METHODS[@]} * ${#SIZES[@]} ))
+ROUND=0
+
+# --parse-only <result-dir>：只解析已有结果目录（${method}_${size}.txt），不执行压测
+if [[ "${1:-}" == "--parse-only" && -n "${2:-}" ]]; then
+  RESULT_DIR="$2"
+fi
+
+LOG_FILE="${RESULT_DIR}/master.log"
+SUMMARY_TSV="${RESULT_DIR}/summary.tsv"
+SUMMARY_MD="${RESULT_DIR}/summary.md"
+
+if [[ "${1:-}" != "--parse-only" ]] && ! command -v warp >/dev/null 2>&1; then
+  echo "错误：未找到 warp 命令，请先安装 MinIO warp。" >&2
+  exit 1
+fi
+
+log() {
+  echo "$(date -u '+%Y-%m-%dT%H:%M:%SZ') $*" | tee -a "${LOG_FILE}"
+}
+
+# ---- 结果解析 ----
+
+# 提取指定 section（GET/PUT/Total 等）的 Average / Reqs / TTFB 原始行
+section_lines() {
+  awk -v sec="$2" '
+    /^Report: / { cur = $2; sub(/\.$/, "", cur) }
+    cur == sec && /^ *\* Average:/ { avg = $0 }
+    cur == sec && /^ *\* Reqs:/   { reqs = $0 }
+    cur == sec && /^ *\* TTFB:/   { ttfb = $0 }
+    END {
+      if (avg != "") print avg
+      if (reqs != "") print reqs
+      if (ttfb != "") print ttfb
+    }
+  ' "$1"
+}
+
+# 从统计行中取字段：tp objs avg p50 p90 p99 ttfb_avg ttfb_p99 ttfb_worst
+field() {
+  case "$2" in
+    tp)         echo "$1" | sed -n 's/^ *\* Average: \(.*\), \([0-9.]*\) obj\/s.*/\1/p' ;;
+    objs)       echo "$1" | sed -n 's/^ *\* Average: .*, \([0-9.]*\) obj\/s.*/\1/p' ;;
+    avg)        echo "$1" | sed -n 's/^ *\* Reqs: Avg: \([^,]*\),.*/\1/p' ;;
+    p50)        echo "$1" | sed -n 's/^ *\* Reqs: Avg: [^,]*, 50%: \([^,]*\),.*/\1/p' ;;
+    p90)        echo "$1" | sed -n 's/^ *\* Reqs: Avg: [^,]*, 50%: [^,]*, 90%: \([^,]*\),.*/\1/p' ;;
+    p99)        echo "$1" | sed -n 's/^ *\* Reqs: Avg: [^,]*, 50%: [^,]*, 90%: [^,]*, 99%: \([^,]*\),.*/\1/p' ;;
+    ttfb_avg)   echo "$1" | sed -n 's/^ *\* TTFB: Avg: \([^,]*\),.*/\1/p' ;;
+    ttfb_p99)   echo "$1" | sed -n 's/^ *\* TTFB: .*99th: \([^,]*\),.*/\1/p' ;;
+    ttfb_worst) echo "$1" | sed -n 's/^ *\* TTFB: .*Worst: \([^ ]*\).*/\1/p' ;;
+    *) echo "" ;;
+  esac
+}
+
+# 解析一轮输出，追加一行到 summary.tsv
+parse_round() {
+  local method="$1" size="$2" file="$3"
+  local line
+  if [[ "${method}" == "mixed" ]]; then
+    local total get put
+    total=$(section_lines "$file" Total)
+    get=$(section_lines "$file" GET)
+    put=$(section_lines "$file" PUT)
+    line=$(printf 'mixed\t%s\t%s\t%s\t%s\t%s' \
+      "$size" \
+      "$(field "${total}" tp)" \
+      "$(field "${total}" objs)" \
+      "$(field "${get}" avg)" \
+      "$(field "${put}" avg)")
+  else
+    local sec
+    sec=$(printf '%s' "${method}" | tr '[:lower:]' '[:upper:]')
+    local stats
+    stats=$(section_lines "$file" "${sec}")
+    line=$(printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s' \
+      "${method}" "${size}" \
+      "$(field "${stats}" tp)" \
+      "$(field "${stats}" objs)" \
+      "$(field "${stats}" avg)" \
+      "$(field "${stats}" p50)" \
+      "$(field "${stats}" p90)" \
+      "$(field "${stats}" p99)" \
+      "$(field "${stats}" ttfb_avg)" \
+      "$(field "${stats}" ttfb_p99)" \
+      "$(field "${stats}" ttfb_worst)")
+  fi
+  printf '%s\n' "${line}" >> "${SUMMARY_TSV}"
+}
+
+# 汇总 summary.tsv -> summary.md（Markdown 表格）
+gen_summary_md() {
+  {
+    echo "# RustFS 性能压测结果"
+    echo ""
+    echo "- 日期：$(date -u '+%Y-%m-%d %H:%M:%S UTC')"
+    echo "- 目标：${HOST}"
+    echo "- 并发：${CONCURRENCY}；单轮：${DURATION}；sleep：${SLEEP_BETWEEN_ROUNDS}s；GET objects：${GET_OBJECTS}"
+    echo "- 方法：GET / PUT / MIXED（warp 默认混合负载）；尺寸：${SIZES[*]}"
+    echo ""
+  } > "${SUMMARY_MD}"
+
+  for m in get put; do
+    {
+      echo "## $(printf '%s' "$m" | tr '[:lower:]' '[:upper:]') 结果"
+      echo ""
+      echo "| 对象尺寸 | 平均吞吐 | 平均 obj/s | Avg Latency | P50 | P90 | P99 | TTFB Avg | TTFB P99 | TTFB 最差 |"
+      echo "|----------|----------|-----------|-------------|-----|-----|-----|----------|----------|-----------|"
+    } >> "${SUMMARY_MD}"
+    while IFS=$'\t' read -r method size tp objs avg p50 p90 p99 ttfb_avg ttfb_p99 ttfb_worst; do
+      [[ "${method}" == "${m}" ]] && \
+        echo "| ${size} | ${tp} | ${objs} | ${avg} | ${p50} | ${p90} | ${p99} | ${ttfb_avg} | ${ttfb_p99} | ${ttfb_worst} |" >> "${SUMMARY_MD}"
+    done < "${SUMMARY_TSV}"
+    echo "" >> "${SUMMARY_MD}"
+  done
+
+  {
+    echo "## MIXED 结果（Total 口径）"
+    echo ""
+    echo "| 对象尺寸 | Total 平均吞吐 | Total 平均 obj/s | Mixed-GET Avg | Mixed-PUT Avg |"
+    echo "|----------|---------------|------------------|----------------|----------------|"
+  } >> "${SUMMARY_MD}"
+  while IFS=$'\t' read -r method size tp objs gavg pavg rest; do
+    [[ "${method}" == "mixed" ]] && \
+      echo "| ${size} | ${tp} | ${objs} | ${gavg} | ${pavg} |" >> "${SUMMARY_MD}"
+  done < "${SUMMARY_TSV}"
+  echo "" >> "${SUMMARY_MD}"
+}
+
+# ---- 主流程 ----
+
+mkdir -p "${RESULT_DIR}"
+log "CONFIG host=${HOST} bucket=${BUCKET} concurrency=${CONCURRENCY} duration=${DURATION} get_objects=${GET_OBJECTS} sleep_between_rounds=${SLEEP_BETWEEN_ROUNDS}s"
+printf 'method\tsize\tthroughput\tobj_per_s\treq_avg\treq_p50\treq_p90\treq_p99\tttfb_avg\tttfb_p99\tttfb_worst\n' > "${SUMMARY_TSV}"
+
+if [[ "${1:-}" == "--parse-only" ]]; then
+  for method in "${METHODS[@]}"; do
+    for size in "${SIZES[@]}"; do
+      outfile="${RESULT_DIR}/${method}_${size}.txt"
+      if [[ -s "${outfile}" ]]; then
+        parse_round "${method}" "${size}" "${outfile}"
+      fi
+    done
+  done
+  gen_summary_md
+  echo "parsed from ${RESULT_DIR}"
+  echo ""
+  cat "${SUMMARY_MD}"
+  exit 0
+fi
+
+for method in "${METHODS[@]}"; do
+  for size in "${SIZES[@]}"; do
+    ROUND=$((ROUND + 1))
+    outfile="${RESULT_DIR}/${method}_${size}.txt"
+
+    log "START round=${ROUND}/${TOTAL_ROUNDS} method=${method} size=${size} concurrency=${CONCURRENCY} duration=${DURATION}"
+
+    extra_args=()
+    if [[ "${method}" != "put" ]]; then
+      extra_args=(--objects "${GET_OBJECTS}")
+    fi
+
+    start_epoch=$(date +%s)
+    warp "${method}" \
+      --host "${HOST}" \
+      --access-key "${ACCESS_KEY}" \
+      --secret-key "${SECRET_KEY}" \
+      --bucket "${BUCKET}" \
+      --concurrent "${CONCURRENCY}" \
+      --duration "${DURATION}" \
+      --obj.size "${size}" \
+      "${extra_args[@]}" \
+      --no-color 2>&1 | tee "${outfile}"
+    rc=${PIPESTATUS[0]}
+    end_epoch=$(date +%s)
+
+    if [[ ${rc} -eq 0 ]]; then
+      parse_round "${method}" "${size}" "${outfile}"
+      log "END round=${ROUND}/${TOTAL_ROUNDS} method=${method} size=${size} rc=${rc} elapsed=$((end_epoch - start_epoch))s parsed=ok"
+    else
+      log "END round=${ROUND}/${TOTAL_ROUNDS} method=${method} size=${size} rc=${rc} elapsed=$((end_epoch - start_epoch))s parsed=skipped"
+    fi
+
+    if [[ ${ROUND} -lt ${TOTAL_ROUNDS} ]]; then
+      log "SLEEP ${SLEEP_BETWEEN_ROUNDS}s before next round"
+      sleep "${SLEEP_BETWEEN_ROUNDS}"
+    fi
+  done
+done
+
+gen_summary_md
+log "ALL_ROUNDS_COMPLETE summary_tsv=${SUMMARY_TSV} summary_md=${SUMMARY_MD}"
+echo ""
+echo "==== 结果汇总 ===="
+cat "${SUMMARY_MD}"
Author	SHA1	Message	Date
overtrue	741d97a118	fix(startup): avoid panic when system CA roots are missing	2026-08-28 05:50:22 +08:00
houseme GitHub heihutu	6f7a4ff060	fix(api): preserve server-side storage error surface (#6753 ) Co-authored-by: heihutu <[email protected]>	2026-08-27 15:42:51 +00:00
hectorandGitHub	8a57632bfd	ci: use dedicated RUSTFS_PERF_NODES for performance test (#6754 )	2026-08-27 22:55:26 +08:00
GatewayJandGitHub	2eb4ddf4af	test(table-catalog): automate DuckDB REST conformance (#6750 ) * test(table-catalog): automate DuckDB REST conformance * fix(table-catalog): protect DuckDB smoke tables	2026-08-27 22:34:10 +08:00
GatewayJandGitHub	e281ed2f6d	test(table-catalog): generate DuckDB REST attach SQL (#6749 )	2026-08-27 22:25:33 +08:00
hectorandGitHub	2e6c820f53	test(heal): relative disk target and fail fast on terminal-but-short (#6748 ) * test(heal): relative disk target and fail fast on terminal-but-short The absolute 40 GiB heal target was calibrated to the background scanner (auto-heal), which is now disabled for determinism; with only the explicit heal the recovered node lands at ~36 GiB for 40 GiB survivors. Make the success criterion relative: the outage node must reach at least 90% of the least-used surviving node (absolute HEAL_TARGET_GB floor optional, default 0 = relative only). Also fail fast when the heal task reaches a terminal success but the disk target is not met (previously the monitor kept polling until timeout), and drop the misleading 'progress absent' warning on the final (cleaned) task response — mid-run progress is reported correctly. Validated live: heal summary=finished, 0 failed, vm000/vm001=40GB, vm002=40GB (target 36GB), test PASSED. * test(heal): gate success on server verdict + data read-back, drop disk GB gate The per-node disk-usage target (40 GiB / 90% of survivors) is not a code-level invariant: EC distributes different shards per node, so the final GB per node depends on the layout, not on heal correctness. Gate the test on what the server actually verifies: - Heal task terminal success (finished/completed) with objectsFailed == 0 (the server's per-object scan/repair verdict). - S3 read-back verification: list the test bucket and GET a sample of objects, requiring HTTP 200 for every read (end-to-end proof the data is still reconstructable after repair). The GET uses a discard mode so binary bodies are not captured (no null-byte warnings / SIGPIPE). Per-node disk usage stays in the output as observability (with a warning if the outage node gained no usage), not as the pass/fail gate. Removes the heal_target_gb input and the relative-target logic. Validated live: heal summary=finished, 0 failed, 20/20 objects read back, vm002_used=40GB, PASS.	2026-08-27 22:25:17 +08:00
hectorandGitHub	d48dda5bdc	ci: add RustFS 4x4 performance test workflow and scripts (#6752 ) * ci: add RustFS 4x4 performance test workflow and scripts * ci: run performance test on dedicated pf-testing runner	2026-08-27 22:24:57 +08:00