shell: use tabwriter for aligned output, remove fragile string error checks

- Use text/tabwriter in runS3UserList and runS3UserAccessKeyList for
  properly aligned column output
- Remove fragile strings.Contains "not found" and "already exists"
  fallbacks in s3ShellIsNotFound and s3ShellIsAlreadyExists; rely on
  typed errors and gRPC status codes only
- Update test assertions to match tabwriter-formatted output

.github/workflows/container_dev.yml

@@ -9,17 +9,96 @@ permissions:
 
 jobs:
 
+  # ── Pre-build Rust volume server binaries natively ──────────────────
+  build-rust-binaries:
+    runs-on: ubuntu-22.04
+    strategy:
+      matrix:
+        include:
+          - target: x86_64-unknown-linux-musl
+            arch: amd64
+          - target: aarch64-unknown-linux-musl
+            arch: arm64
+            cross: true
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+
+      - name: Install protobuf compiler
+        run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          targets: ${{ matrix.target }}
+
+      - name: Install musl tools (amd64)
+        if: ${{ !matrix.cross }}
+        run: sudo apt-get install -y musl-tools
+
+      - name: Install cross-compilation tools (arm64)
+        if: matrix.cross
+        run: |
+          sudo apt-get install -y gcc-aarch64-linux-gnu
+          echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER=aarch64-linux-gnu-gcc" >> "$GITHUB_ENV"
+          # Disable glibc fortify source — its __memcpy_chk etc. symbols don't exist in musl
+          echo "CFLAGS_aarch64_unknown_linux_musl=-U_FORTIFY_SOURCE" >> "$GITHUB_ENV"
+
+      - name: Cache cargo registry and target
+        uses: actions/cache@v5
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            seaweed-volume/target
+          key: rust-docker-dev-${{ matrix.target }}-${{ hashFiles('seaweed-volume/Cargo.lock') }}
+          restore-keys: |
+            rust-docker-dev-${{ matrix.target }}-
+
+      - name: Build normal variant
+        env:
+          SEAWEEDFS_COMMIT: ${{ github.sha }}
+        run: |
+          cd seaweed-volume
+          cargo build --release --target ${{ matrix.target }} --no-default-features
+          cp target/${{ matrix.target }}/release/weed-volume ../weed-volume-normal-${{ matrix.arch }}
+
+      - name: Upload artifacts
+        uses: actions/upload-artifact@v7
+        with:
+          name: rust-volume-${{ matrix.arch }}
+          path: weed-volume-normal-${{ matrix.arch }}
+
   build-dev-containers:
+    needs: [build-rust-binaries]
     runs-on: [ubuntu-latest]
 
     steps:
-      -
-        name: Checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v2
-      -
-        name: Docker meta
+      - name: Checkout
+        uses: actions/checkout@v6
+
+      - name: Download pre-built Rust binaries
+        uses: actions/download-artifact@v8
+        with:
+          pattern: rust-volume-*
+          merge-multiple: true
+          path: ./rust-bins
+
+      - name: Place Rust binaries in Docker context
+        run: |
+          mkdir -p docker/weed-volume-prebuilt
+          for arch in amd64 arm64; do
+            src="./rust-bins/weed-volume-normal-${arch}"
+            if [ -f "$src" ]; then
+              cp "$src" "docker/weed-volume-prebuilt/weed-volume-${arch}"
+              echo "Placed pre-built Rust binary for ${arch}"
+            fi
+          done
+          ls -la docker/weed-volume-prebuilt/
+
+      - name: Docker meta
         id: docker_meta
-        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v3
+        uses: docker/metadata-action@v6
         with:
           images: |
             chrislusf/seaweedfs
@@ -30,40 +109,40 @@ jobs:
             org.opencontainers.image.title=seaweedfs
             org.opencontainers.image.description=SeaweedFS is a distributed storage system for blobs, objects, files, and data lake, to store and serve billions of files fast!
             org.opencontainers.image.vendor=Chris Lu
-      -
-        name: Set up QEMU
-        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v1
-      -
-        name: Create BuildKit config
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v4
+
+      - name: Create BuildKit config
         run: |
           cat > /tmp/buildkitd.toml <<EOF
           [registry."docker.io"]
             mirrors = ["https://mirror.gcr.io"]
           EOF
-      -
-        name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v1
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v4
         with:
           buildkitd-flags: "--debug"
           buildkitd-config: /tmp/buildkitd.toml
-      -
-        name: Login to Docker Hub
+
+      - name: Login to Docker Hub
         if: github.event_name != 'pull_request'
-        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
-      -
-        name: Login to GHCR
+
+      - name: Login to GHCR
         if: github.event_name != 'pull_request'
-        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
+        uses: docker/login-action@v4.1.0
         with:
           registry: ghcr.io
           username: ${{ secrets.GHCR_USERNAME }}
           password: ${{ secrets.GHCR_TOKEN }}
-      -
-        name: Build
-        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v2
+
+      - name: Build
+        uses: docker/build-push-action@v7
         with:
           context: ./docker
           push: ${{ github.event_name != 'pull_request' }}

.github/workflows/container_foundationdb_version.yml

@@ -133,7 +133,7 @@ jobs:
 
       - name: Login to Docker Hub
         if: github.event_name != 'pull_request'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}

.github/workflows/container_latest.yml

@@ -58,8 +58,80 @@ jobs:
             echo "publish=true" >> "$GITHUB_OUTPUT"
           fi
 
+  # ── Pre-build Rust volume server binaries natively ──────────────────
+  build-rust-binaries:
+    runs-on: ubuntu-22.04
+    strategy:
+      matrix:
+        include:
+          - target: x86_64-unknown-linux-musl
+            arch: amd64
+          - target: aarch64-unknown-linux-musl
+            arch: arm64
+            cross: true
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          ref: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.source_ref || github.ref }}
+
+      - name: Install protobuf compiler
+        run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          targets: ${{ matrix.target }}
+
+      - name: Install musl tools (amd64)
+        if: ${{ !matrix.cross }}
+        run: sudo apt-get install -y musl-tools
+
+      - name: Install cross-compilation tools (arm64)
+        if: matrix.cross
+        run: |
+          sudo apt-get install -y gcc-aarch64-linux-gnu
+          echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER=aarch64-linux-gnu-gcc" >> "$GITHUB_ENV"
+          # Disable glibc fortify source — its __memcpy_chk etc. symbols don't exist in musl
+          echo "CFLAGS_aarch64_unknown_linux_musl=-U_FORTIFY_SOURCE" >> "$GITHUB_ENV"
+
+      - name: Cache cargo registry and target
+        uses: actions/cache@v5
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            seaweed-volume/target
+          key: rust-docker-${{ matrix.target }}-${{ hashFiles('seaweed-volume/Cargo.lock') }}
+          restore-keys: |
+            rust-docker-${{ matrix.target }}-
+
+      - name: Build large-disk variant
+        env:
+          SEAWEEDFS_COMMIT: ${{ github.sha }}
+        run: |
+          cd seaweed-volume
+          cargo build --release --target ${{ matrix.target }}
+          cp target/${{ matrix.target }}/release/weed-volume ../weed-volume-large-disk-${{ matrix.arch }}
+
+      - name: Build normal variant
+        env:
+          SEAWEEDFS_COMMIT: ${{ github.sha }}
+        run: |
+          cd seaweed-volume
+          cargo build --release --target ${{ matrix.target }} --no-default-features
+          cp target/${{ matrix.target }}/release/weed-volume ../weed-volume-normal-${{ matrix.arch }}
+
+      - name: Upload artifacts
+        uses: actions/upload-artifact@v7
+        with:
+          name: rust-volume-${{ matrix.arch }}
+          path: |
+            weed-volume-large-disk-${{ matrix.arch }}
+            weed-volume-normal-${{ matrix.arch }}
+
   build:
-    needs: [setup]
+    needs: [setup, build-rust-binaries]
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -94,11 +166,32 @@ jobs:
           if [ "${{ matrix.variant }}" == "large_disk" ]; then
             echo "tag_suffix=_large_disk" >> $GITHUB_OUTPUT
             echo "build_args=TAGS=5BytesOffset" >> $GITHUB_OUTPUT
+            echo "rust_variant=large-disk" >> $GITHUB_OUTPUT
           else
             echo "tag_suffix=" >> $GITHUB_OUTPUT
             echo "build_args=" >> $GITHUB_OUTPUT
+            echo "rust_variant=normal" >> $GITHUB_OUTPUT
           fi
 
+      - name: Download pre-built Rust binaries
+        uses: actions/download-artifact@v8
+        with:
+          pattern: rust-volume-*
+          merge-multiple: true
+          path: ./rust-bins
+
+      - name: Place Rust binaries in Docker context
+        run: |
+          mkdir -p docker/weed-volume-prebuilt
+          for arch in amd64 arm64; do
+            src="./rust-bins/weed-volume-${{ steps.config.outputs.rust_variant }}-${arch}"
+            if [ -f "$src" ]; then
+              cp "$src" "docker/weed-volume-prebuilt/weed-volume-${arch}"
+              echo "Placed pre-built Rust binary for ${arch}"
+            fi
+          done
+          ls -la docker/weed-volume-prebuilt/
+
       - name: Docker meta
         id: docker_meta
         uses: docker/metadata-action@v6
@@ -127,13 +220,13 @@ jobs:
           buildkitd-config: /tmp/buildkitd.toml
       - name: Login to Docker Hub
         if: needs.setup.outputs.publish == 'true'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Login to GHCR
         if: needs.setup.outputs.publish == 'true'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           registry: ghcr.io
           username: ${{ secrets.GHCR_USERNAME }}
@@ -166,7 +259,7 @@ jobs:
 
   trivy-scan:
     runs-on: ubuntu-latest
-    needs: [setup, build]
+    needs: [setup, build, build-rust-binaries]
     strategy:
       matrix:
         variant: ${{ fromJSON(needs.setup.outputs.variants) }}
@@ -181,7 +274,7 @@ jobs:
           fi
       - name: Login to GHCR
         if: needs.setup.outputs.publish == 'true'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           registry: ghcr.io
           username: ${{ secrets.GHCR_USERNAME }}
@@ -191,6 +284,29 @@ jobs:
         uses: actions/checkout@v6
         with:
           ref: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.source_ref || github.ref }}
+      - name: Download pre-built Rust binaries for local scan
+        if: needs.setup.outputs.publish != 'true'
+        uses: actions/download-artifact@v8
+        with:
+          pattern: rust-volume-*
+          merge-multiple: true
+          path: ./rust-bins
+      - name: Place Rust binaries in Docker context for local scan
+        if: needs.setup.outputs.publish != 'true'
+        run: |
+          rust_variant="normal"
+          if [ "${{ matrix.variant }}" == "large_disk" ]; then
+            rust_variant="large-disk"
+          fi
+          mkdir -p docker/weed-volume-prebuilt
+          for arch in amd64 arm64; do
+            src="./rust-bins/weed-volume-${rust_variant}-${arch}"
+            if [ -f "$src" ]; then
+              cp "$src" "docker/weed-volume-prebuilt/weed-volume-${arch}"
+              echo "Placed pre-built Rust binary for ${arch}"
+            fi
+          done
+          ls -la docker/weed-volume-prebuilt/
       - name: Create BuildKit config for local scan build
         if: needs.setup.outputs.publish != 'true'
         run: |
@@ -313,12 +429,12 @@ jobs:
             ghcr.io/chrislusf/seaweedfs
           tags: type=raw,value=${{ github.event_name == 'workflow_dispatch' && github.event.inputs.image_tag || 'latest' }},suffix=${{ steps.config.outputs.tag_suffix }}
       - name: Login to Docker Hub
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Login to GHCR
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           registry: ghcr.io
           username: ${{ secrets.GHCR_USERNAME }}

.github/workflows/container_release_foundationdb.yml

@@ -42,7 +42,7 @@ jobs:
       -
         name: Login to Docker Hub
         if: github.event_name != 'pull_request'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}

.github/workflows/container_release_unified.yml

@@ -74,6 +74,8 @@ jobs:
         run: |
           sudo apt-get install -y gcc-aarch64-linux-gnu
           echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER=aarch64-linux-gnu-gcc" >> "$GITHUB_ENV"
+          # Disable glibc fortify source — its __memcpy_chk etc. symbols don't exist in musl
+          echo "CFLAGS_aarch64_unknown_linux_musl=-U_FORTIFY_SOURCE" >> "$GITHUB_ENV"
 
       - name: Cache cargo registry and target
         uses: actions/cache@v5
@@ -103,7 +105,7 @@ jobs:
           cp target/${{ matrix.target }}/release/weed-volume ../weed-volume-normal-${{ matrix.arch }}
 
       - name: Upload artifacts
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         with:
           name: rust-volume-${{ matrix.arch }}
           path: |
@@ -166,7 +168,7 @@ jobs:
 
       - name: Download pre-built Rust binaries
         if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v8
         with:
           pattern: rust-volume-*
           merge-multiple: true
@@ -234,14 +236,14 @@ jobs:
           
       - name: Login to Docker Hub
         if: (github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant) && github.event_name != 'pull_request'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
           
       - name: Login to GHCR
         if: (github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant) && github.event_name != 'pull_request'
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           registry: ghcr.io
           username: ${{ secrets.GHCR_USERNAME }}
@@ -297,14 +299,14 @@ jobs:
     steps:
       - name: Login to Docker Hub
         if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
           
       - name: Login to GHCR
         if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
-        uses: docker/login-action@v4
+        uses: docker/login-action@v4.1.0
         with:
           registry: ghcr.io
           username: ${{ secrets.GHCR_USERNAME }}

.github/workflows/container_rocksdb_version.yml

@@ -88,7 +88,7 @@ jobs:
         uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v1
 
       - name: Login to Docker Hub
-        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
+        uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v1
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}

.github/workflows/rust_binaries_release.yml

@@ -99,7 +99,7 @@ jobs:
 
       - name: Upload artifacts
         if: ${{ !startsWith(github.ref, 'refs/tags/') }}
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         with:
           name: rust-volume-${{ matrix.asset_suffix }}
           path: |
@@ -176,7 +176,7 @@ jobs:
 
       - name: Upload artifacts
         if: ${{ !startsWith(github.ref, 'refs/tags/') }}
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         with:
           name: rust-volume-${{ matrix.asset_suffix }}
           path: |
@@ -245,7 +245,7 @@ jobs:
 
       - name: Upload artifacts
         if: ${{ !startsWith(github.ref, 'refs/tags/') }}
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         with:
           name: rust-volume-windows_amd64
           path: |

go.mod

@@ -34,7 +34,7 @@ require (
 	github.com/google/btree v1.1.3
 	github.com/google/uuid v1.6.0
 	github.com/google/wire v0.7.0 // indirect
-	github.com/googleapis/gax-go/v2 v2.17.0 // indirect
+	github.com/googleapis/gax-go/v2 v2.19.0 // indirect
 	github.com/gorilla/mux v1.8.1
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -86,7 +86,7 @@ require (
 	github.com/xdg-go/scram v1.2.0
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
-	go.etcd.io/etcd/client/v3 v3.6.9
+	go.etcd.io/etcd/client/v3 v3.6.10
 	go.mongodb.org/mongo-driver v1.17.9
 	go.opencensus.io v0.24.0 // indirect
 	gocloud.dev v0.45.0
@@ -95,22 +95,21 @@ require (
 	golang.org/x/crypto v0.49.0
 	golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa
 	golang.org/x/image v0.38.0
-	golang.org/x/net v0.51.0
+	golang.org/x/net v0.52.0
 	golang.org/x/oauth2 v0.36.0
 	golang.org/x/sys v0.42.0
 	golang.org/x/text v0.35.0 // indirect
 	golang.org/x/tools v0.42.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
-	google.golang.org/api v0.267.0
-	google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 // indirect
+	google.golang.org/api v0.274.0
+	google.golang.org/genproto v0.0.0-20260316180232-0b37fe3546d5 // indirect
 	google.golang.org/grpc v1.79.3
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	modernc.org/b v1.0.0 // indirect
-	modernc.org/mathutil v1.7.1
+	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect
 	modernc.org/sqlite v1.46.1
-	modernc.org/strutil v1.2.1
 )
 
 require (
@@ -126,8 +125,8 @@ require (
 	github.com/armon/go-metrics v0.4.1
 	github.com/aws/aws-sdk-go-v2 v1.41.5
 	github.com/aws/aws-sdk-go-v2/config v1.32.13
-	github.com/aws/aws-sdk-go-v2/credentials v1.19.13
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.14
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0
 	github.com/cognusion/imaging v1.0.2
 	github.com/fluent/fluent-logger-golang v1.10.1
 	github.com/getsentry/sentry-go v0.44.1
@@ -136,10 +135,9 @@ require (
 	github.com/google/flatbuffers/go v0.0.0-20230108230133-3b8644d32c50
 	github.com/hashicorp/raft v1.7.3
 	github.com/hashicorp/raft-boltdb/v2 v2.3.1
-	github.com/hashicorp/vault/api v1.22.0
+	github.com/hashicorp/vault/api v1.23.0
 	github.com/jhump/protoreflect v1.18.0
 	github.com/linkedin/goavro/v2 v2.15.0
-	github.com/mattn/go-sqlite3 v1.14.34
 	github.com/minio/crc64nvme v1.1.1
 	github.com/orcaman/concurrent-map/v2 v2.0.1
 	github.com/parquet-go/parquet-go v0.28.0
@@ -157,7 +155,7 @@ require (
 	github.com/xeipuuv/gojsonschema v1.2.0
 	github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.1
 	github.com/ydb-platform/ydb-go-sdk/v3 v3.125.3
-	go.etcd.io/etcd/client/pkg/v3 v3.6.9
+	go.etcd.io/etcd/client/pkg/v3 v3.6.10
 	go.uber.org/atomic v1.11.0
 	golang.org/x/sync v0.20.0
 	golang.org/x/tools/godoc v0.1.0-deprecated
@@ -291,7 +289,7 @@ require (
 
 require (
 	cel.dev/expr v0.25.1 // indirect
-	cloud.google.com/go/auth v0.18.1 // indirect
+	cloud.google.com/go/auth v0.18.2 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
 	cloud.google.com/go/compute/metadata v0.9.0 // indirect
 	cloud.google.com/go/iam v1.5.3 // indirect
@@ -323,21 +321,21 @@ require (
 	github.com/andybalholm/cascadia v1.3.3 // indirect
 	github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc // indirect
 	github.com/arangodb/go-velocypack v0.0.0-20200318135517-5af53c29c67e // indirect
-	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.12 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sns v1.39.7 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sqs v1.42.17 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.30.14 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.18 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.30.15 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.19 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.10
 	github.com/aws/smithy-go v1.24.2
 	github.com/boltdb/bolt v1.3.1 // indirect
@@ -387,7 +385,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.5.2 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
-	github.com/googleapis/enterprise-certificate-proxy v0.3.11 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect
 	github.com/gorilla/schema v1.4.1 // indirect
 	github.com/gorilla/securecookie v1.1.2 // indirect
 	github.com/gorilla/sessions v1.4.0
@@ -486,22 +484,22 @@ require (
 	github.com/zeebo/blake3 v0.2.4 // indirect
 	github.com/zeebo/errs v1.4.0 // indirect
 	go.etcd.io/bbolt v1.4.3 // indirect
-	go.etcd.io/etcd/api/v3 v3.6.9 // indirect
+	go.etcd.io/etcd/api/v3 v3.6.10 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.39.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 // indirect
-	go.opentelemetry.io/otel v1.40.0 // indirect
-	go.opentelemetry.io/otel/metric v1.40.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.40.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.40.0 // indirect
-	go.opentelemetry.io/otel/trace v1.40.0 // indirect
+	go.opentelemetry.io/otel v1.42.0 // indirect
+	go.opentelemetry.io/otel/metric v1.42.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.42.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.42.0 // indirect
+	go.opentelemetry.io/otel/trace v1.42.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.1 // indirect
 	golang.org/x/term v0.41.0 // indirect
-	golang.org/x/time v0.14.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20 // indirect
+	golang.org/x/time v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260316180232-0b37fe3546d5 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260319201613-d00831a3d3e7 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/validator.v2 v2.0.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

go.sum

@@ -94,8 +94,8 @@ cloud.google.com/go/assuredworkloads v1.7.0/go.mod h1:z/736/oNmtGAyU47reJgGN+KVo
 cloud.google.com/go/assuredworkloads v1.8.0/go.mod h1:AsX2cqyNCOvEQC8RMPnoc0yEarXQk6WEKkxYfL6kGIo=
 cloud.google.com/go/assuredworkloads v1.9.0/go.mod h1:kFuI1P78bplYtT77Tb1hi0FMxM0vVpRC7VVoJC3ZoT0=
 cloud.google.com/go/assuredworkloads v1.10.0/go.mod h1:kwdUQuXcedVdsIaKgKTp9t0UJkE5+PAVNhdQm4ZVq2E=
-cloud.google.com/go/auth v0.18.1 h1:IwTEx92GFUo2pJ6Qea0EU3zYvKnTAeRCODxfA/G5UWs=
-cloud.google.com/go/auth v0.18.1/go.mod h1:GfTYoS9G3CWpRA3Va9doKN9mjPGRS+v41jmZAhBzbrA=
+cloud.google.com/go/auth v0.18.2 h1:+Nbt5Ev0xEqxlNjd6c+yYUeosQ5TtEUaNcN/3FozlaM=
+cloud.google.com/go/auth v0.18.2/go.mod h1:xD+oY7gcahcu7G2SG2DsBerfFxgPAJz17zz2joOFF3M=
 cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc=
 cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c=
 cloud.google.com/go/automl v1.5.0/go.mod h1:34EjfoFGMZ5sgJ9EoLsRtdPSNZLcfflJR39VbVNS2M0=
@@ -310,8 +310,8 @@ cloud.google.com/go/lifesciences v0.6.0/go.mod h1:ddj6tSX/7BOnhxCSd3ZcETvtNr8NZ6
 cloud.google.com/go/lifesciences v0.8.0/go.mod h1:lFxiEOMqII6XggGbOnKiyZ7IBwoIqA84ClvoezaA/bo=
 cloud.google.com/go/logging v1.6.1/go.mod h1:5ZO0mHHbvm8gEmeEUHrmDlTDSu5imF6MUP9OfilNXBw=
 cloud.google.com/go/logging v1.7.0/go.mod h1:3xjP2CjkM3ZkO73aj4ASA5wRPGGCRrPIAeNqVNkzY8M=
-cloud.google.com/go/logging v1.13.1 h1:O7LvmO0kGLaHY/gq8cV7T0dyp6zJhYAOtZPX4TF3QtY=
-cloud.google.com/go/logging v1.13.1/go.mod h1:XAQkfkMBxQRjQek96WLPNze7vsOmay9H5PqfsNYDqvw=
+cloud.google.com/go/logging v1.13.2 h1:qqlHCBvieJT9Cdq4QqYx1KPadCQ2noD4FK02eNqHAjA=
+cloud.google.com/go/logging v1.13.2/go.mod h1:zaybliM3yun1J8mU2dVQ1/qDzjbOqEijZCn6hSBtKak=
 cloud.google.com/go/longrunning v0.1.1/go.mod h1:UUFxuDWkv22EuY93jjmDMFT5GPQKeFVJBIF6QlTqdsE=
 cloud.google.com/go/longrunning v0.3.0/go.mod h1:qth9Y41RRSUE69rDcOn6DdK3HfQfsUI0YSmW3iIlLJc=
 cloud.google.com/go/longrunning v0.4.1/go.mod h1:4iWDqhBZ70CvZ6BfETbvam3T8FMvLK+eFj0E6AaRQTo=
@@ -716,12 +716,12 @@ github.com/aws/aws-sdk-go v1.55.8 h1:JRmEUbU52aJQZ2AjX4q4Wu7t4uZjOu71uyNmaWlUkJQ
 github.com/aws/aws-sdk-go v1.55.8/go.mod h1:ZkViS9AqA6otK+JBBNH2++sx1sgxrPKcSzPPvQkUtXk=
 github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY=
 github.com/aws/aws-sdk-go-v2 v1.41.5/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o=
-github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU=
-github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4/go.mod h1:IOAPF6oT9KCsceNTvvYMNHy0+kMF8akOjeDvPENWxp4=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 h1:eBMB84YGghSocM7PsjmmPffTa+1FBUeNvGvFou6V/4o=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8/go.mod h1:lyw7GFp3qENLh7kwzf7iMzAxDn+NzjXEAGjKS2UOKqI=
 github.com/aws/aws-sdk-go-v2/config v1.32.13 h1:5KgbxMaS2coSWRrx9TX/QtWbqzgQkOdEa3sZPhBhCSg=
 github.com/aws/aws-sdk-go-v2/config v1.32.13/go.mod h1:8zz7wedqtCbw5e9Mi2doEwDyEgHcEE9YOJp6a8jdSMY=
-github.com/aws/aws-sdk-go-v2/credentials v1.19.13 h1:mA59E3fokBvyEGHKFdnpNNrvaR351cqiHgRg+JzOSRI=
-github.com/aws/aws-sdk-go-v2/credentials v1.19.13/go.mod h1:yoTXOQKea18nrM69wGF9jBdG4WocSZA1h38A+t/MAsk=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.14 h1:n+UcGWAIZHkXzYt87uMFBv/l8THYELoX6gVcUvgl6fI=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.14/go.mod h1:cJKuyWB59Mqi0jM3nFYQRmnHVQIcgoxjEMAbLkpr62w=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21 h1:NUS3K4BTDArQqNu2ih7yeDLaS3bmHD0YndtA6UP884g=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21/go.mod h1:YWNWJQNjKigKY1RHVJCuupeWDrrHjRqHm0N9rdrWzYI=
 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.12 h1:Zy6Tme1AA13kX8x3CnkHx5cqdGWGaj/anwOiWGnA0Xo=
@@ -732,28 +732,28 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 h1:PEgGVtPoB6NTpPrBgq
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21/go.mod h1:p+hz+PRAYlY3zcpJhPwXlLC4C+kqn70WIHwnzAfs6ps=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 h1:qYQ4pzQ2Oz6WpQ8T3HvGHnZydA72MnLuFK9tJwmrbHw=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6/go.mod h1:O3h0IK87yXci+kg6flUKzJnWeziQUKciKrLjcatSNcY=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17 h1:JqcdRG//czea7Ppjb+g/n4o8i/R50aTBHkA7vu0lK+k=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17/go.mod h1:CO+WeGmIdj/MlPel2KwID9Gt7CNq4M65HUfBW97liM0=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 h1:rWyie/PxDRIdhNf4DzRk0lvjVOqFJuNnO8WwaIRVxzQ=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22/go.mod h1:zd/JsJ4P7oGfUhXn1VyLqaRZwPmZwg44Jf2dS84Dm3Y=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 h1:5EniKhLZe4xzL7a+fU3C2tfUN4nWIqlLesfrjkuPFTY=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7/go.mod h1:x0nZssQ3qZSnIcePWLvcoFisRXJzcTVvYpAAdYX8+GI=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8 h1:Z5EiPIzXKewUQK0QTMkutjiaPVeVYXX7KIqhXu/0fXs=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8/go.mod h1:FsTpJtvC4U1fyDXk7c71XoDv3HlRm8V3NiYLeYLh5YE=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 h1:JRaIgADQS/U6uXDqlPiefP32yXTda7Kqfx+LgspooZM=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13/go.mod h1:CEuVn5WqOMilYl+tbccq8+N2ieCy0gVn3OtRb0vBNNM=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3xgIJMSC8S6hEVq+38DcvUlgFY0FM6mSI5oto=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21/go.mod h1:r6+pf23ouCB718FUxaqzZdbpYFyDtehyZcmP5KL9FkA=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 h1:bGeHBsGZx0Dvu/eJC0Lh9adJa3M1xREcndxLNZlve2U=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17/go.mod h1:dcW24lbU0CzHusTE8LLHhRLI42ejmINN8Lcr22bwh/g=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0 h1:oeu8VPlOre74lBA/PMhxa5vewaMIMmILM+RraSyB8KA=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0/go.mod h1:5jggDlZ2CLQhwJBiZJb4vfk4f0GxWdEDruWKEJ1xOdo=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 h1:ZlvrNcHSFFWURB8avufQq9gFsheUgjVD9536obIknfM=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21/go.mod h1:cv3TNhVrssKR0O/xxLJVRfd2oazSnZnkUeTf6ctUwfQ=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0 h1:foqo/ocQ7WqKwy3FojGtZQJo0FR4vto9qnz9VaumbCo=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0/go.mod h1:uoA43SdFwacedBfSgfFSjjCvYe8aYBS7EnU5GZ/YKMM=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 h1:QKZH0S178gCmFEgst8hN0mCX1KxLgHBKKY/CLqwP8lg=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.9/go.mod h1:7yuQJoT+OoH8aqIxw9vwF+8KpvLZ8AWmvmUWHsGQZvI=
 github.com/aws/aws-sdk-go-v2/service/sns v1.39.7 h1:fovS7qGMT+BBSuifkySdVaMWxXTyaYT6qaBx/1y6Ij4=
 github.com/aws/aws-sdk-go-v2/service/sns v1.39.7/go.mod h1:gFahrattA8ulEtiS4XL/fQiQ77l+Urc52Y96/r1e6ks=
 github.com/aws/aws-sdk-go-v2/service/sqs v1.42.17 h1:ZNMxVFPayuHe14u/vn+BwLi3wxQvxcNTw8WdPv2gqBc=
 github.com/aws/aws-sdk-go-v2/service/sqs v1.42.17/go.mod h1:ZxqweFQ2w6NNznWMUvWV9AvkAfM6J8F/MC250Mb4n1I=
-github.com/aws/aws-sdk-go-v2/service/sso v1.30.14 h1:GcLE9ba5ehAQma6wlopUesYg/hbcOhFNWTjELkiWkh4=
-github.com/aws/aws-sdk-go-v2/service/sso v1.30.14/go.mod h1:WSvS1NLr7JaPunCXqpJnWk1Bjo7IxzZXrZi1QQCkuqM=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.18 h1:mP49nTpfKtpXLt5SLn8Uv8z6W+03jYVoOSAl/c02nog=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.18/go.mod h1:YO8TrYtFdl5w/4vmjL8zaBSsiNp3w0L1FfKVKenZT7w=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.15 h1:lFd1+ZSEYJZYvv9d6kXzhkZu07si3f+GQ1AaYwa2LUM=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.15/go.mod h1:WSvS1NLr7JaPunCXqpJnWk1Bjo7IxzZXrZi1QQCkuqM=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.19 h1:dzztQ1YmfPrxdrOiuZRMF6fuOwWlWpD2StNLTceKpys=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.19/go.mod h1:YO8TrYtFdl5w/4vmjL8zaBSsiNp3w0L1FfKVKenZT7w=
 github.com/aws/aws-sdk-go-v2/service/sts v1.41.10 h1:p8ogvvLugcR/zLBXTXrTkj0RYBUdErbMnAFFp12Lm/U=
 github.com/aws/aws-sdk-go-v2/service/sts v1.41.10/go.mod h1:60dv0eZJfeVXfbT1tFJinbHrDfSJ2GZl4Q//OSSNAVw=
 github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng=
@@ -1284,8 +1284,8 @@ github.com/googleapis/enterprise-certificate-proxy v0.1.0/go.mod h1:17drOmN3MwGY
 github.com/googleapis/enterprise-certificate-proxy v0.2.0/go.mod h1:8C0jb7/mgJe/9KK8Lm7X9ctZC2t60YyIpYEI16jx0Qg=
 github.com/googleapis/enterprise-certificate-proxy v0.2.1/go.mod h1:AwSRAtLfXpU5Nm3pW+v7rGDHp09LsPtGY9MduiEsR9k=
 github.com/googleapis/enterprise-certificate-proxy v0.2.3/go.mod h1:AwSRAtLfXpU5Nm3pW+v7rGDHp09LsPtGY9MduiEsR9k=
-github.com/googleapis/enterprise-certificate-proxy v0.3.11 h1:vAe81Msw+8tKUxi2Dqh/NZMz7475yUvmRIkXr4oN2ao=
-github.com/googleapis/enterprise-certificate-proxy v0.3.11/go.mod h1:RFV7MUdlb7AgEq2v7FmMCfeSMCllAzWxFgRdusoGks8=
+github.com/googleapis/enterprise-certificate-proxy v0.3.14 h1:yh8ncqsbUY4shRD5dA6RlzjJaT4hi3kII+zYw8wmLb8=
+github.com/googleapis/enterprise-certificate-proxy v0.3.14/go.mod h1:vqVt9yG9480NtzREnTlmGSBmFrA+bzb0yl0TxoBQXOg=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gax-go/v2 v2.1.0/go.mod h1:Q3nei7sK6ybPYH7twZdmQpAd1MKb7pfu6SK+H1/DsU0=
@@ -1296,8 +1296,8 @@ github.com/googleapis/gax-go/v2 v2.4.0/go.mod h1:XOTVJ59hdnfJLIP/dh8n5CGryZR2LxK
 github.com/googleapis/gax-go/v2 v2.5.1/go.mod h1:h6B0KMMFNtI2ddbGJn3T3ZbwkeT6yqEF02fYlzkUCyo=
 github.com/googleapis/gax-go/v2 v2.6.0/go.mod h1:1mjbznJAPHFpesgE5ucqfYEscaz5kMdcIDwU/6+DDoY=
 github.com/googleapis/gax-go/v2 v2.7.0/go.mod h1:TEop28CZZQ2y+c0VxMUmu1lV+fQx57QpBWsYpwqHJx8=
-github.com/googleapis/gax-go/v2 v2.17.0 h1:RksgfBpxqff0EZkDWYuz9q/uWsTVz+kf43LsZ1J6SMc=
-github.com/googleapis/gax-go/v2 v2.17.0/go.mod h1:mzaqghpQp4JDh3HvADwrat+6M3MOIDp5YKHhb9PAgDY=
+github.com/googleapis/gax-go/v2 v2.19.0 h1:fYQaUOiGwll0cGj7jmHT/0nPlcrZDFPrZRhTsoCr8hE=
+github.com/googleapis/gax-go/v2 v2.19.0/go.mod h1:w2ROXVdfGEVFXzmlciUU4EdjHgWvB5h2n6x/8XSTTJA=
 github.com/googleapis/go-type-adapters v1.0.0/go.mod h1:zHW75FOG2aur7gAO2B+MLby+cLsWGBF62rFAi7WjWO4=
 github.com/googleapis/google-cloud-go-testing v0.0.0-20200911160855-bcd43fbb19e8/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g=
 github.com/gookit/color v1.4.2/go.mod h1:fqRyamkC1W8uxl+lxCQxOT09l/vYfZ+QeiX3rKQHCoQ=
@@ -1383,8 +1383,8 @@ github.com/hashicorp/raft-boltdb v0.0.0-20230125174641-2a8082862702 h1:RLKEcCuKc
 github.com/hashicorp/raft-boltdb v0.0.0-20230125174641-2a8082862702/go.mod h1:nTakvJ4XYq45UXtn0DbwR4aU9ZdjlnIenpbs6Cd+FM0=
 github.com/hashicorp/raft-boltdb/v2 v2.3.1 h1:ackhdCNPKblmOhjEU9+4lHSJYFkJd6Jqyvj6eW9pwkc=
 github.com/hashicorp/raft-boltdb/v2 v2.3.1/go.mod h1:n4S+g43dXF1tqDT+yzcXHhXM6y7MrlUd3TTwGRcUvQE=
-github.com/hashicorp/vault/api v1.22.0 h1:+HYFquE35/B74fHoIeXlZIP2YADVboaPjaSicHEZiH0=
-github.com/hashicorp/vault/api v1.22.0/go.mod h1:IUZA2cDvr4Ok3+NtK2Oq/r+lJeXkeCrHRmqdyWfpmGM=
+github.com/hashicorp/vault/api v1.23.0 h1:gXgluBsSECfRWTSW9niY2jwg2e9mMJc4WoHNv4g3h6A=
+github.com/hashicorp/vault/api v1.23.0/go.mod h1:zransKiB9ftp+kgY8ydjnvCU7Wk8i9L0DYWpXeMj9ko=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
 github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
@@ -2101,12 +2101,12 @@ go.einride.tech/aip v0.73.0 h1:bPo4oqBo2ZQeBKo4ZzLb1kxYXTY1ysJhpvQyfuGzvps=
 go.einride.tech/aip v0.73.0/go.mod h1:Mj7rFbmXEgw0dq1dqJ7JGMvYCZZVxmGOR3S4ZcV5LvQ=
 go.etcd.io/bbolt v1.4.3 h1:dEadXpI6G79deX5prL3QRNP6JB8UxVkqo4UPnHaNXJo=
 go.etcd.io/bbolt v1.4.3/go.mod h1:tKQlpPaYCVFctUIgFKFnAlvbmB3tpy1vkTnDWohtc0E=
-go.etcd.io/etcd/api/v3 v3.6.9 h1:UA7iKfEW1AzgihcBSGXci2kDGQiokSq41F9HMCI/RTI=
-go.etcd.io/etcd/api/v3 v3.6.9/go.mod h1:csEk/qTfxKL36NqJdU15Tgtl65A8dyEY2BYo7PRsIwk=
-go.etcd.io/etcd/client/pkg/v3 v3.6.9 h1:T8nuk8Lz64C+Hzb0coBFLMSlVSQZBpAtFk46swdM1DA=
-go.etcd.io/etcd/client/pkg/v3 v3.6.9/go.mod h1:WEy3PpwbbEBVRdh1NVJYsuUe/8eyI21PNJRazeD8z/Y=
-go.etcd.io/etcd/client/v3 v3.6.9 h1:3X555hQXmhRr27O37wls53g68CpUiPOiHXrZfz2Al+o=
-go.etcd.io/etcd/client/v3 v3.6.9/go.mod h1:KO7H1HLYh1qaljuVZJQwBFk1lRce6pJzt+C81GEnrlM=
+go.etcd.io/etcd/api/v3 v3.6.10 h1:jlwjtELjA8yi2VWpOFH+0w0lGr3K6mVDyn0RDB9aaAY=
+go.etcd.io/etcd/api/v3 v3.6.10/go.mod h1:pdV4VeFmvhdNjB4LWRkC8ReLyRBAxUOze3GarMhE2sk=
+go.etcd.io/etcd/client/pkg/v3 v3.6.10 h1:tBT7podcPhuVbCVkAEzx8bC5I+aqxfLwBN8/As1arrA=
+go.etcd.io/etcd/client/pkg/v3 v3.6.10/go.mod h1:WEy3PpwbbEBVRdh1NVJYsuUe/8eyI21PNJRazeD8z/Y=
+go.etcd.io/etcd/client/v3 v3.6.10 h1:J598zJ+C/ZPvImypmq5waj84+bovePrlZERHklf34y0=
+go.etcd.io/etcd/client/v3 v3.6.10/go.mod h1:iHhUDUcEwaKs1YFq3MgmI9U4zhTVasp/vgdVbFf1RS8=
 go.mongodb.org/mongo-driver v1.17.9 h1:IexDdCuuNJ3BHrELgBlyaH9p60JXAvdzWR128q+U5tU=
 go.mongodb.org/mongo-driver v1.17.9/go.mod h1:LlOhpH5NUEfhxcAwG0UEkMqwYcc4JU18gtCdGudk/tQ=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
@@ -2128,8 +2128,8 @@ go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.
 go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.60.0/go.mod h1:CosX/aS4eHnG9D7nESYpV753l4j9q5j3SL/PUYd2lR8=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 h1:RbKq8BG0FI8OiXhBfcRtqqHcZcka+gU3cskNuf05R18=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0/go.mod h1:h06DGIukJOevXaj/xrNjhi/2098RZzcLTbc0jDAUbsg=
-go.opentelemetry.io/otel v1.40.0 h1:oA5YeOcpRTXq6NN7frwmwFR0Cn3RhTVZvXsP4duvCms=
-go.opentelemetry.io/otel v1.40.0/go.mod h1:IMb+uXZUKkMXdPddhwAHm6UfOwJyh4ct1ybIlV14J0g=
+go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho=
+go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.38.0 h1:vl9obrcoWVKp/lwl8tRE33853I8Xru9HFbw/skNeLs8=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.38.0/go.mod h1:GAXRxmLJcVM3u22IjTg74zWBrRCKq8BnOqUVLodpcpw=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.35.0 h1:0NIXxOCFx+SKbhCVxwl3ETG8ClLPAa0KuKV6p3yhxP8=
@@ -2144,14 +2144,14 @@ go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.39.0 h1:5gn2urDL/FBnK8
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.39.0/go.mod h1:0fBG6ZJxhqByfFZDwSwpZGzJU671HkwpWaNe2t4VUPI=
 go.opentelemetry.io/otel/exporters/zipkin v1.36.0 h1:s0n95ya5tOG03exJ5JySOdJFtwGo4ZQ+KeY7Zro4CLI=
 go.opentelemetry.io/otel/exporters/zipkin v1.36.0/go.mod h1:m9wRxtKA2MZ1HcnNC4BKI+9aYe434qRZTCvI7QGUN7Y=
-go.opentelemetry.io/otel/metric v1.40.0 h1:rcZe317KPftE2rstWIBitCdVp89A2HqjkxR3c11+p9g=
-go.opentelemetry.io/otel/metric v1.40.0/go.mod h1:ib/crwQH7N3r5kfiBZQbwrTge743UDc7DTFVZrrXnqc=
-go.opentelemetry.io/otel/sdk v1.40.0 h1:KHW/jUzgo6wsPh9At46+h4upjtccTmuZCFAc9OJ71f8=
-go.opentelemetry.io/otel/sdk v1.40.0/go.mod h1:Ph7EFdYvxq72Y8Li9q8KebuYUr2KoeyHx0DRMKrYBUE=
-go.opentelemetry.io/otel/sdk/metric v1.40.0 h1:mtmdVqgQkeRxHgRv4qhyJduP3fYJRMX4AtAlbuWdCYw=
-go.opentelemetry.io/otel/sdk/metric v1.40.0/go.mod h1:4Z2bGMf0KSK3uRjlczMOeMhKU2rhUqdWNoKcYrtcBPg=
-go.opentelemetry.io/otel/trace v1.40.0 h1:WA4etStDttCSYuhwvEa8OP8I5EWu24lkOzp+ZYblVjw=
-go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA=
+go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4=
+go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI=
+go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo=
+go.opentelemetry.io/otel/sdk v1.42.0/go.mod h1:rGHCAxd9DAph0joO4W6OPwxjNTYWghRWmkHuGbayMts=
+go.opentelemetry.io/otel/sdk/metric v1.42.0 h1:D/1QR46Clz6ajyZ3G8SgNlTJKBdGp84q9RKCAZ3YGuA=
+go.opentelemetry.io/otel/sdk/metric v1.42.0/go.mod h1:Ua6AAlDKdZ7tdvaQKfSmnFTdHx37+J4ba8MwVCYM5hc=
+go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY=
+go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.opentelemetry.io/proto/otlp v0.15.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
 go.opentelemetry.io/proto/otlp v0.19.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
@@ -2344,8 +2344,8 @@ golang.org/x/net v0.16.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
-golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
-golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
+golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -2552,8 +2552,8 @@ golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxb
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.1.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
-golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -2703,8 +2703,8 @@ google.golang.org/api v0.106.0/go.mod h1:2Ts0XTHNVWxypznxWOYUeI4g3WdP9Pk2Qk58+a/
 google.golang.org/api v0.107.0/go.mod h1:2Ts0XTHNVWxypznxWOYUeI4g3WdP9Pk2Qk58+a/O9MY=
 google.golang.org/api v0.108.0/go.mod h1:2Ts0XTHNVWxypznxWOYUeI4g3WdP9Pk2Qk58+a/O9MY=
 google.golang.org/api v0.110.0/go.mod h1:7FC4Vvx1Mooxh8C5HWjzZHcavuS2f6pmJpZx60ca7iI=
-google.golang.org/api v0.267.0 h1:w+vfWPMPYeRs8qH1aYYsFX68jMls5acWl/jocfLomwE=
-google.golang.org/api v0.267.0/go.mod h1:Jzc0+ZfLnyvXma3UtaTl023TdhZu6OMBP9tJ+0EmFD0=
+google.golang.org/api v0.274.0 h1:aYhycS5QQCwxHLwfEHRRLf9yNsfvp1JadKKWBE54RFA=
+google.golang.org/api v0.274.0/go.mod h1:JbAt7mF+XVmWu6xNP8/+CTiGH30ofmCmk9nM8d8fHew=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -2838,12 +2838,12 @@ google.golang.org/genproto v0.0.0-20230209215440-0dfe4f8abfcc/go.mod h1:RGgjbofJ
 google.golang.org/genproto v0.0.0-20230216225411-c8e22ba71e44/go.mod h1:8B0gmkoRebU8ukX6HP+4wrVQUY1+6PkQ44BSyIlflHA=
 google.golang.org/genproto v0.0.0-20230222225845-10f96fb3dbec/go.mod h1:3Dl5ZL0q0isWJt+FVcfpQyirqemEuLAK/iFvg1UP1Hw=
 google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4/go.mod h1:NWraEVixdDnqcqQ30jipen1STv2r/n24Wb7twVTGR4s=
-google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 h1:VQZ/yAbAtjkHgH80teYd2em3xtIkkHd7ZhqfH2N9CsM=
-google.golang.org/genproto v0.0.0-20260128011058-8636f8732409/go.mod h1:rxKD3IEILWEu3P44seeNOAwZN4SaoKaQ/2eTg4mM6EM=
-google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20 h1:7ei4lp52gK1uSejlA8AZl5AJjeLUOHBQscRQZUgAcu0=
-google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20/go.mod h1:ZdbssH/1SOVnjnDlXzxDHK2MCidiqXtbYccJNzNYPEE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20 h1:Jr5R2J6F6qWyzINc+4AM8t5pfUz6beZpHp678GNrMbE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
+google.golang.org/genproto v0.0.0-20260316180232-0b37fe3546d5 h1:JNfk58HZ8lfmXbYK2vx/UvsqIL59TzByCxPIX4TDmsE=
+google.golang.org/genproto v0.0.0-20260316180232-0b37fe3546d5/go.mod h1:x5julN69+ED4PcFk/XWayw35O0lf/nGa4aNgODCmNmw=
+google.golang.org/genproto/googleapis/api v0.0.0-20260316180232-0b37fe3546d5 h1:CogIeEXn4qWYzzQU0QqvYBM8yDF9cFYzDq9ojSpv0Js=
+google.golang.org/genproto/googleapis/api v0.0.0-20260316180232-0b37fe3546d5/go.mod h1:EIQZ5bFCfRQDV4MhRle7+OgjNtZ6P1PiZBgAKuxXu/Y=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260319201613-d00831a3d3e7 h1:ndE4FoJqsIceKP2oYSnUZqhTdYufCYYkqwtFzfrhI7w=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260319201613-d00831a3d3e7/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=

seaweed-volume/Cargo.toml

@@ -82,7 +82,7 @@ memmap2 = "0.9"
 uuid = { version = "1", features = ["v4"] }
 
 # HTTP client (for proxying, remote fetch)
-reqwest = { version = "0.12", features = ["rustls-tls", "stream", "multipart", "json"] }
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream", "multipart", "json"] }
 
 # Content hashing
 md-5 = "0.10"

weed/admin/dash/config_persistence.go

@@ -1091,7 +1091,7 @@ func (cp *ConfigPersistence) loadTaskStateLocked(taskID string) (*maintenance.Ma
 	// Convert protobuf to maintenance task
 	task := cp.protobufToMaintenanceTask(taskStateFile.Task)
 
-	glog.V(2).Infof("Loaded task state for task %s from %s", taskID, taskFilePath)
+	glog.V(3).Infof("Loaded task state for task %s from %s", taskID, taskFilePath)
 	return task, nil
 }
 
@@ -1135,6 +1135,43 @@ func (cp *ConfigPersistence) loadAllTaskStatesLocked() ([]*maintenance.Maintenan
 	return tasks, nil
 }
 
+// DeleteAllTaskStates removes all task state .pb files from disk without reading them.
+// Used at startup to clean up stale files from previous runs — the scanner will
+// re-detect any tasks that are still needed from live cluster state.
+func (cp *ConfigPersistence) DeleteAllTaskStates() error {
+	cp.tasksMu.Lock()
+	defer cp.tasksMu.Unlock()
+
+	if cp.dataDir == "" {
+		return nil
+	}
+
+	tasksDir := filepath.Join(cp.dataDir, TasksSubdir)
+	entries, err := os.ReadDir(tasksDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return fmt.Errorf("failed to read tasks directory: %w", err)
+	}
+
+	var removed int
+	for _, entry := range entries {
+		if !entry.IsDir() && filepath.Ext(entry.Name()) == ".pb" {
+			if err := os.Remove(filepath.Join(tasksDir, entry.Name())); err != nil && !os.IsNotExist(err) {
+				glog.Warningf("Failed to delete task file %s: %v", entry.Name(), err)
+			} else {
+				removed++
+			}
+		}
+	}
+
+	if removed > 0 {
+		glog.Infof("Cleaned up %d stale task files from disk", removed)
+	}
+	return nil
+}
+
 // DeleteTaskState removes a task state file from disk
 func (cp *ConfigPersistence) DeleteTaskState(taskID string) error {
 	cp.tasksMu.Lock()

weed/admin/maintenance/maintenance_queue.go

@@ -33,80 +33,16 @@ func (mq *MaintenanceQueue) SetPersistence(persistence TaskPersistence) {
 	glog.V(1).Infof("Maintenance queue configured with task persistence")
 }
 
-// LoadTasksFromPersistence loads tasks from persistent storage on startup
+// LoadTasksFromPersistence is called on startup. Previous task states are NOT loaded
+// into memory — the maintenance scanner will re-detect current needs from the live
+// cluster state. Stale task files from previous runs are deleted from disk.
 func (mq *MaintenanceQueue) LoadTasksFromPersistence() error {
-	if mq.persistence == nil {
-		glog.V(1).Infof("No task persistence configured, skipping task loading")
-		return nil
-	}
-
-	mq.mutex.Lock()
-	defer mq.mutex.Unlock()
-
-	glog.Infof("Loading tasks from persistence...")
-
-	tasks, err := mq.persistence.LoadAllTaskStates()
-	if err != nil {
-		return fmt.Errorf("failed to load task states: %w", err)
-	}
-
-	glog.Infof("DEBUG LoadTasksFromPersistence: Found %d tasks in persistence", len(tasks))
-
-	// Reset task maps
-	mq.tasks = make(map[string]*MaintenanceTask)
-	mq.pendingTasks = make([]*MaintenanceTask, 0)
-
-	// Load tasks by status
-	for _, task := range tasks {
-		glog.Infof("DEBUG LoadTasksFromPersistence: Loading task %s (type: %s, status: %s, scheduled: %v)", task.ID, task.Type, task.Status, task.ScheduledAt)
-		mq.tasks[task.ID] = task
-
-		switch task.Status {
-		case TaskStatusPending:
-			glog.Infof("DEBUG LoadTasksFromPersistence: Adding task %s to pending queue", task.ID)
-			mq.pendingTasks = append(mq.pendingTasks, task)
-		case TaskStatusAssigned, TaskStatusInProgress:
-			// For assigned/in-progress tasks, we need to check if the worker is still available
-			// If not, we should fail them and make them eligible for retry
-			if task.WorkerID != "" {
-				if _, exists := mq.workers[task.WorkerID]; !exists {
-					glog.Warningf("Task %s was assigned to unavailable worker %s, marking as failed", task.ID, task.WorkerID)
-					task.Status = TaskStatusFailed
-					task.Error = "Worker unavailable after restart"
-					completedTime := time.Now()
-					task.CompletedAt = &completedTime
-
-					// Check if it should be retried
-					if task.RetryCount < task.MaxRetries {
-						task.RetryCount++
-						task.Status = TaskStatusPending
-						task.WorkerID = ""
-						task.StartedAt = nil
-						task.CompletedAt = nil
-						task.Error = ""
-						task.ScheduledAt = time.Now().Add(1 * time.Minute) // Retry after restart delay
-						glog.Infof("DEBUG LoadTasksFromPersistence: Retrying task %s, adding to pending queue", task.ID)
-						mq.pendingTasks = append(mq.pendingTasks, task)
-					}
-				}
-			}
-		}
-
-		// Sync task with ActiveTopology for capacity tracking
-		if mq.integration != nil {
-			mq.integration.SyncTask(task)
+	if mq.persistence != nil {
+		if err := mq.persistence.DeleteAllTaskStates(); err != nil {
+			glog.Warningf("Failed to clean up old task files: %v", err)
 		}
 	}
-
-	// Sort pending tasks by priority and schedule time
-	sort.Slice(mq.pendingTasks, func(i, j int) bool {
-		if mq.pendingTasks[i].Priority != mq.pendingTasks[j].Priority {
-			return mq.pendingTasks[i].Priority > mq.pendingTasks[j].Priority
-		}
-		return mq.pendingTasks[i].ScheduledAt.Before(mq.pendingTasks[j].ScheduledAt)
-	})
-
-	glog.Infof("Loaded %d tasks from persistence (%d pending)", len(tasks), len(mq.pendingTasks))
+	glog.Infof("Task queue initialized (previous tasks will be re-detected by scanner)")
 	return nil
 }
 
@@ -119,6 +55,14 @@ func (mq *MaintenanceQueue) saveTaskState(task *MaintenanceTask) {
 	}
 }
 
+func (mq *MaintenanceQueue) deleteTaskState(taskID string) {
+	if mq.persistence != nil {
+		if err := mq.persistence.DeleteTaskState(taskID); err != nil {
+			glog.V(2).Infof("Failed to delete task state for %s: %v", taskID, err)
+		}
+	}
+}
+
 // cleanupCompletedTasks removes old completed tasks beyond the retention limit
 func (mq *MaintenanceQueue) cleanupCompletedTasks() {
 	if mq.persistence != nil {
@@ -128,10 +72,24 @@ func (mq *MaintenanceQueue) cleanupCompletedTasks() {
 	}
 }
 
+const MaxTasksPerType = 100
+
 // AddTask adds a new maintenance task to the queue with deduplication
 func (mq *MaintenanceQueue) AddTask(task *MaintenanceTask) {
 	mq.mutex.Lock()
 
+	// Enforce per-type capacity limit (only counting active tasks)
+	if mq.countActiveTasksByType(task.Type) >= MaxTasksPerType {
+		// Purge terminal tasks first, then recheck
+		mq.purgeTerminalTasksLocked()
+		if mq.countActiveTasksByType(task.Type) >= MaxTasksPerType {
+			mq.mutex.Unlock()
+			glog.V(1).Infof("Task skipped (type %s at capacity %d): volume %d on %s",
+				task.Type, MaxTasksPerType, task.VolumeID, task.Server)
+			return
+		}
+	}
+
 	// Enforce one queued/active task per volume (across all task types).
 	if mq.hasQueuedOrActiveTaskForVolume(task.VolumeID) {
 		mq.mutex.Unlock()
@@ -197,6 +155,30 @@ func (mq *MaintenanceQueue) AddTask(task *MaintenanceTask) {
 		taskSnapshot.ID, taskSnapshot.Type, taskSnapshot.VolumeID, taskSnapshot.Server, taskSnapshot.Priority, scheduleInfo, taskSnapshot.Reason)
 }
 
+// countActiveTasksByType returns the number of active (non-terminal) tasks of a given type. Caller must hold mq.mutex.
+func (mq *MaintenanceQueue) countActiveTasksByType(taskType MaintenanceTaskType) int {
+	count := 0
+	for _, t := range mq.tasks {
+		if t.Type == taskType {
+			switch t.Status {
+			case TaskStatusPending, TaskStatusAssigned, TaskStatusInProgress:
+				count++
+			}
+		}
+	}
+	return count
+}
+
+// purgeTerminalTasksLocked removes terminal tasks from the in-memory map. Caller must hold mq.mutex.
+func (mq *MaintenanceQueue) purgeTerminalTasksLocked() {
+	for id, task := range mq.tasks {
+		switch task.Status {
+		case TaskStatusCompleted, TaskStatusFailed, TaskStatusCancelled:
+			delete(mq.tasks, id)
+		}
+	}
+}
+
 // hasQueuedOrActiveTaskForVolume checks if any pending/assigned/in-progress task already exists for this volume.
 // Caller must hold mq.mutex.
 func (mq *MaintenanceQueue) hasQueuedOrActiveTaskForVolume(volumeID uint32) bool {
@@ -273,6 +255,9 @@ func (mq *MaintenanceQueue) CancelPendingTasksByType(taskType MaintenanceTaskTyp
 
 // AddTasksFromResults converts detection results to tasks and adds them to the queue
 func (mq *MaintenanceQueue) AddTasksFromResults(results []*TaskDetectionResult) {
+	// Purge terminal tasks from memory before adding new ones
+	mq.purgeTerminalTasks()
+
 	for _, result := range results {
 		// Validate that task has proper typed parameters
 		if result.TypedParams == nil {
@@ -297,6 +282,21 @@ func (mq *MaintenanceQueue) AddTasksFromResults(results []*TaskDetectionResult)
 	}
 }
 
+// purgeTerminalTasks removes completed/failed/cancelled tasks from memory.
+// Terminal tasks are already deleted from disk by CompleteTask, so this
+// only needs to clean up the in-memory map.
+func (mq *MaintenanceQueue) purgeTerminalTasks() {
+	mq.mutex.Lock()
+	before := len(mq.tasks)
+	mq.purgeTerminalTasksLocked()
+	purged := before - len(mq.tasks)
+	mq.mutex.Unlock()
+
+	if purged > 0 {
+		glog.V(1).Infof("Purged %d terminal tasks from memory", purged)
+	}
+}
+
 // GetNextTask returns the next available task for a worker
 func (mq *MaintenanceQueue) GetNextTask(workerID string, capabilities []MaintenanceTaskType) *MaintenanceTask {
 	// Use read lock for initial checks and search
@@ -570,7 +570,6 @@ func (mq *MaintenanceQueue) CompleteTask(taskID string, error string) {
 		}
 	}
 	taskStatus := task.Status
-	taskCount := len(mq.tasks)
 	// Snapshot task state while lock is still held to avoid data race
 	var taskToSaveSnapshot *MaintenanceTask
 	if taskToSave != nil {
@@ -578,9 +577,18 @@ func (mq *MaintenanceQueue) CompleteTask(taskID string, error string) {
 	}
 	mq.mutex.Unlock()
 
-	// Save task state to persistence outside the lock
+	// Only persist non-terminal tasks (retries). Completed/failed tasks stay
+	// in memory for the UI but are not written to disk — they would just
+	// accumulate and slow down future startups.
 	if taskToSaveSnapshot != nil {
-		mq.saveTaskState(taskToSaveSnapshot)
+		switch taskStatus {
+		case TaskStatusPending:
+			// Retry — save so the task survives a restart
+			mq.saveTaskState(taskToSaveSnapshot)
+		case TaskStatusCompleted, TaskStatusFailed, TaskStatusCancelled:
+			// Terminal — delete the file if one exists from a previous state
+			mq.deleteTaskState(taskToSaveSnapshot.ID)
+		}
 	}
 
 	if logFn != nil {
@@ -591,13 +599,6 @@ func (mq *MaintenanceQueue) CompleteTask(taskID string, error string) {
 	if taskStatus != TaskStatusPending {
 		mq.removePendingOperation(taskID)
 	}
-
-	// Periodically cleanup old completed tasks (when total task count is a multiple of 10)
-	if taskStatus == TaskStatusCompleted {
-		if taskCount%10 == 0 {
-			go mq.cleanupCompletedTasks()
-		}
-	}
 }
 
 // isNonRetriableError returns true for errors that will never succeed on retry,

weed/admin/maintenance/maintenance_queue_test.go

@@ -693,10 +693,11 @@ func (m *MockPersistence) SaveTaskState(task *MaintenanceTask) error
 func (m *MockPersistence) LoadTaskState(taskID string) (*MaintenanceTask, error)    { return nil, nil }
 func (m *MockPersistence) LoadAllTaskStates() ([]*MaintenanceTask, error)           { return m.tasks, nil }
 func (m *MockPersistence) DeleteTaskState(taskID string) error                      { return nil }
+func (m *MockPersistence) DeleteAllTaskStates() error                               { return nil }
 func (m *MockPersistence) CleanupCompletedTasks() error                             { return nil }
 func (m *MockPersistence) SaveTaskPolicy(taskType string, policy *TaskPolicy) error { return nil }
 
-func TestMaintenanceQueue_LoadTasksCapacitySync(t *testing.T) {
+func TestMaintenanceQueue_LoadTasksStartsEmpty(t *testing.T) {
 	// Setup
 	policy := &MaintenancePolicy{
 		TaskPolicies: map[string]*worker_pb.TaskPolicy{
@@ -704,56 +705,25 @@ func TestMaintenanceQueue_LoadTasksCapacitySync(t *testing.T) {
 		},
 	}
 	mq := NewMaintenanceQueue(policy)
-	integration := NewMaintenanceIntegration(mq, policy)
-	mq.SetIntegration(integration)
-	at := integration.GetActiveTopology()
 
-	topologyInfo := &master_pb.TopologyInfo{
-		DataCenterInfos: []*master_pb.DataCenterInfo{
-			{
-				Id: "dc1",
-				RackInfos: []*master_pb.RackInfo{
-					{
-						Id: "rack1",
-						DataNodeInfos: []*master_pb.DataNodeInfo{
-							{
-								Id: "server1",
-								DiskInfos: map[string]*master_pb.DiskInfo{
-									"hdd1": {DiskId: 1, VolumeCount: 1, MaxVolumeCount: 10},
-									"hdd2": {DiskId: 2, VolumeCount: 0, MaxVolumeCount: 10},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-	at.UpdateTopology(topologyInfo)
-
-	// Setup mock persistence with a pending task
-	taskID := "load_test_123"
+	// Setup mock persistence with tasks — these should NOT be loaded
 	mockTask := &MaintenanceTask{
-		ID:     taskID,
+		ID:     "old_task_123",
 		Type:   "balance",
 		Status: TaskStatusPending,
-		TypedParams: &worker_pb.TaskParams{
-			TaskId:  taskID,
-			Sources: []*worker_pb.TaskSource{{Node: "server1", DiskId: 1}},
-			Targets: []*worker_pb.TaskTarget{{Node: "server1", DiskId: 2}},
-		},
 	}
 	mq.SetPersistence(&MockPersistence{tasks: []*MaintenanceTask{mockTask}})
 
-	// Load tasks
+	// LoadTasksFromPersistence should be a no-op — scanner will re-detect
 	err := mq.LoadTasksFromPersistence()
 	if err != nil {
-		t.Fatalf("Failed to load tasks: %v", err)
+		t.Fatalf("LoadTasksFromPersistence failed: %v", err)
 	}
 
-	// Verify capacity is reserved in ActiveTopology after loading (9 left)
-	if at.GetEffectiveAvailableCapacity("server1", 2) != 9 {
-		t.Errorf("Expected capacity 9 after loading tasks, got %d", at.GetEffectiveAvailableCapacity("server1", 2))
+	// Queue should be empty — tasks will be re-detected by scanner
+	stats := mq.GetStats()
+	if stats.TotalTasks != 0 {
+		t.Errorf("Expected 0 tasks after startup, got %d", stats.TotalTasks)
 	}
 }
 

weed/admin/maintenance/maintenance_scanner.go

@@ -32,21 +32,18 @@ func NewMaintenanceScanner(adminClient AdminClient, policy *MaintenancePolicy, q
 
 // ScanForMaintenanceTasks analyzes the cluster and generates maintenance tasks
 func (ms *MaintenanceScanner) ScanForMaintenanceTasks() ([]*TaskDetectionResult, error) {
-	// Get volume health metrics
-	volumeMetrics, err := ms.getVolumeHealthMetrics()
+	// Get volume health metrics directly in task-system format, along with topology info
+	taskMetrics, topologyInfo, err := ms.getVolumeHealthMetrics()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get volume health metrics: %w", err)
 	}
 
 	// Use task system for all task types
 	if ms.integration != nil {
-		// Convert metrics to task system format
-		taskMetrics := ms.convertToTaskMetrics(volumeMetrics)
-
 		// Update topology information for complete cluster view (including empty servers)
 		// This must happen before task detection to ensure EC placement can consider all servers
-		if ms.lastTopologyInfo != nil {
-			if err := ms.integration.UpdateTopologyInfo(ms.lastTopologyInfo); err != nil {
+		if topologyInfo != nil {
+			if err := ms.integration.UpdateTopologyInfo(topologyInfo); err != nil {
 				glog.Errorf("Failed to update topology info for empty servers: %v", err)
 				// Don't fail the scan - continue with just volume-bearing servers
 			} else {
@@ -70,9 +67,12 @@ func (ms *MaintenanceScanner) ScanForMaintenanceTasks() ([]*TaskDetectionResult,
 	return []*TaskDetectionResult{}, nil
 }
 
-// getVolumeHealthMetrics collects health information for all volumes
-func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*VolumeHealthMetrics, error) {
-	var metrics []*VolumeHealthMetrics
+// getVolumeHealthMetrics collects health information for all volumes.
+// Returns metrics in task-system format directly (no intermediate copy) and
+// the topology info for updating the active topology.
+func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*types.VolumeHealthMetrics, *master_pb.TopologyInfo, error) {
+	var metrics []*types.VolumeHealthMetrics
+	var topologyInfo *master_pb.TopologyInfo
 
 	glog.V(1).Infof("Collecting volume health metrics from master")
 	err := ms.adminClient.WithMasterClient(func(client master_pb.SeaweedClient) error {
@@ -89,30 +89,28 @@ func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*VolumeHealthMetrics,
 
 		volumeSizeLimitBytes := uint64(resp.VolumeSizeLimitMb) * 1024 * 1024 // Convert MB to bytes
 
-		// Track all nodes discovered in topology
-		var allNodesInTopology []string
-		var nodesWithVolumes []string
-		var nodesWithoutVolumes []string
+		// Track node counts for summary logging (avoid accumulating full ID slices)
+		var totalNodes, nodesWithVolumes, nodesWithoutVolumes int
 
 		for _, dc := range resp.TopologyInfo.DataCenterInfos {
-			glog.V(2).Infof("Processing datacenter: %s", dc.Id)
+			glog.V(3).Infof("Processing datacenter: %s", dc.Id)
 			for _, rack := range dc.RackInfos {
-				glog.V(2).Infof("Processing rack: %s in datacenter: %s", rack.Id, dc.Id)
+				glog.V(3).Infof("Processing rack: %s in datacenter: %s", rack.Id, dc.Id)
 				for _, node := range rack.DataNodeInfos {
-					allNodesInTopology = append(allNodesInTopology, node.Id)
-					glog.V(2).Infof("Found volume server in topology: %s (disks: %d)", node.Id, len(node.DiskInfos))
+					totalNodes++
+					glog.V(3).Infof("Found volume server in topology: %s (disks: %d)", node.Id, len(node.DiskInfos))
 
 					hasVolumes := false
 					// Process each disk on this node
 					for diskType, diskInfo := range node.DiskInfos {
 						if len(diskInfo.VolumeInfos) > 0 {
 							hasVolumes = true
-							glog.V(2).Infof("Volume server %s disk %s has %d volumes", node.Id, diskType, len(diskInfo.VolumeInfos))
+							glog.V(3).Infof("Volume server %s disk %s has %d volumes", node.Id, diskType, len(diskInfo.VolumeInfos))
 						}
 
 						// Process volumes on this specific disk
 						for _, volInfo := range diskInfo.VolumeInfos {
-							metric := &VolumeHealthMetrics{
+							metric := &types.VolumeHealthMetrics{
 								VolumeID:         volInfo.Id,
 								Server:           node.Id,
 								ServerAddress:    node.Address,
@@ -138,7 +136,7 @@ func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*VolumeHealthMetrics,
 							}
 							metric.Age = time.Since(metric.LastModified)
 
-							glog.V(3).Infof("Volume %d on %s:%s (ID %d): size=%d, limit=%d, fullness=%.2f",
+							glog.V(4).Infof("Volume %d on %s:%s (ID %d): size=%d, limit=%d, fullness=%.2f",
 								metric.VolumeID, metric.Server, metric.DiskType, metric.DiskId, metric.Size, volumeSizeLimitBytes, metric.FullnessRatio)
 
 							metrics = append(metrics, metric)
@@ -146,29 +144,27 @@ func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*VolumeHealthMetrics,
 					}
 
 					if hasVolumes {
-						nodesWithVolumes = append(nodesWithVolumes, node.Id)
+						nodesWithVolumes++
 					} else {
-						nodesWithoutVolumes = append(nodesWithoutVolumes, node.Id)
+						nodesWithoutVolumes++
 						glog.V(1).Infof("Volume server %s found in topology but has no volumes", node.Id)
 					}
 				}
 			}
 		}
 
-		glog.Infof("Topology discovery complete:")
-		glog.Infof("  - Total volume servers in topology: %d (%v)", len(allNodesInTopology), allNodesInTopology)
-		glog.Infof("  - Volume servers with volumes: %d (%v)", len(nodesWithVolumes), nodesWithVolumes)
-		glog.Infof("  - Volume servers without volumes: %d (%v)", len(nodesWithoutVolumes), nodesWithoutVolumes)
+		glog.Infof("Topology discovery: %d volume servers (%d with volumes, %d without)",
+			totalNodes, nodesWithVolumes, nodesWithoutVolumes)
 
-		// Store topology info for volume shard tracker
-		ms.lastTopologyInfo = resp.TopologyInfo
+		// Return topology info as a local value (not retained on the scanner struct)
+		topologyInfo = resp.TopologyInfo
 
 		return nil
 	})
 
 	if err != nil {
 		glog.Errorf("Failed to get volume health metrics: %v", err)
-		return nil, err
+		return nil, nil, err
 	}
 
 	glog.V(1).Infof("Successfully collected metrics for %d actual volumes with disk ID information", len(metrics))
@@ -176,13 +172,13 @@ func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*VolumeHealthMetrics,
 	// Count actual replicas and identify EC volumes
 	ms.enrichVolumeMetrics(metrics)
 
-	return metrics, nil
+	return metrics, topologyInfo, nil
 }
 
 // enrichVolumeMetrics adds additional information like replica counts
-func (ms *MaintenanceScanner) enrichVolumeMetrics(metrics []*VolumeHealthMetrics) {
+func (ms *MaintenanceScanner) enrichVolumeMetrics(metrics []*types.VolumeHealthMetrics) {
 	// Group volumes by ID to count replicas
-	volumeGroups := make(map[uint32][]*VolumeHealthMetrics)
+	volumeGroups := make(map[uint32][]*types.VolumeHealthMetrics)
 	for _, metric := range metrics {
 		volumeGroups[metric.VolumeID] = append(volumeGroups[metric.VolumeID], metric)
 	}
@@ -193,41 +189,9 @@ func (ms *MaintenanceScanner) enrichVolumeMetrics(metrics []*VolumeHealthMetrics
 		for _, replica := range replicas {
 			replica.ReplicaCount = replicaCount
 		}
-		glog.V(3).Infof("Volume %d has %d replicas", volumeID, replicaCount)
+		glog.V(4).Infof("Volume %d has %d replicas", volumeID, replicaCount)
 	}
 
 	// TODO: Identify EC volumes by checking volume structure
 	// This would require querying volume servers for EC shard information
 }
-
-// convertToTaskMetrics converts existing volume metrics to task system format
-func (ms *MaintenanceScanner) convertToTaskMetrics(metrics []*VolumeHealthMetrics) []*types.VolumeHealthMetrics {
-	var simplified []*types.VolumeHealthMetrics
-
-	for _, metric := range metrics {
-		simplified = append(simplified, &types.VolumeHealthMetrics{
-			VolumeID:         metric.VolumeID,
-			Server:           metric.Server,
-			ServerAddress:    metric.ServerAddress,
-			DiskType:         metric.DiskType,
-			DiskId:           metric.DiskId,
-			DataCenter:       metric.DataCenter,
-			Rack:             metric.Rack,
-			Collection:       metric.Collection,
-			Size:             metric.Size,
-			DeletedBytes:     metric.DeletedBytes,
-			GarbageRatio:     metric.GarbageRatio,
-			LastModified:     metric.LastModified,
-			Age:              metric.Age,
-			ReplicaCount:     metric.ReplicaCount,
-			ExpectedReplicas: metric.ExpectedReplicas,
-			IsReadOnly:       metric.IsReadOnly,
-			HasRemoteCopy:    metric.HasRemoteCopy,
-			IsECVolume:       metric.IsECVolume,
-			FullnessRatio:    metric.FullnessRatio,
-		})
-	}
-
-	glog.V(2).Infof("Converted %d volume metrics with disk ID information for task detection", len(simplified))
-	return simplified
-}

weed/admin/maintenance/maintenance_types.go

@@ -130,6 +130,7 @@ type TaskPersistence interface {
 	LoadTaskState(taskID string) (*MaintenanceTask, error)
 	LoadAllTaskStates() ([]*MaintenanceTask, error)
 	DeleteTaskState(taskID string) error
+	DeleteAllTaskStates() error
 	CleanupCompletedTasks() error
 
 	// Policy persistence
@@ -206,12 +207,11 @@ type MaintenanceQueue struct {
 
 // MaintenanceScanner analyzes the cluster and generates maintenance tasks
 type MaintenanceScanner struct {
-	adminClient      AdminClient
-	policy           *MaintenancePolicy
-	queue            *MaintenanceQueue
-	lastScan         map[MaintenanceTaskType]time.Time
-	integration      *MaintenanceIntegration
-	lastTopologyInfo *master_pb.TopologyInfo
+	adminClient AdminClient
+	policy      *MaintenancePolicy
+	queue       *MaintenanceQueue
+	lastScan    map[MaintenanceTaskType]time.Time
+	integration *MaintenanceIntegration
 }
 
 // TaskDetectionResult represents the result of scanning for maintenance needs

weed/admin/topology/topology_management.go

@@ -87,7 +87,7 @@ func (at *ActiveTopology) UpdateTopology(topologyInfo *master_pb.TopologyInfo) e
 					}
 
 					diskKey := fmt.Sprintf("%s:%d", nodeInfo.Id, diskInfo.DiskId)
-					glog.V(2).Infof("UpdateTopology: adding disk key=%q nodeId=%q diskId=%d diskType=%q address=%q grpcPort=%d volumes=%d maxVolumes=%d",
+					glog.V(3).Infof("UpdateTopology: adding disk key=%q nodeId=%q diskId=%d diskType=%q address=%q grpcPort=%d volumes=%d maxVolumes=%d",
 						diskKey, nodeInfo.Id, diskInfo.DiskId, diskType, nodeInfo.Address, nodeInfo.GrpcPort, diskInfo.VolumeCount, diskInfo.MaxVolumeCount)
 					node.disks[diskInfo.DiskId] = disk
 					at.disks[diskKey] = disk

weed/admin/view/app/object_store_users_templ.go

@@ -41,7 +41,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 		var templ_7745c5c3_Var2 string
 		templ_7745c5c3_Var2, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", data.TotalUsers))
 		if templ_7745c5c3_Err != nil {
-			return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 38, Col: 71}
+			return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 38, Col: 71}
 		}
 		_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var2))
 		if templ_7745c5c3_Err != nil {
@@ -54,7 +54,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 		var templ_7745c5c3_Var3 string
 		templ_7745c5c3_Var3, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.Users)))
 		if templ_7745c5c3_Err != nil {
-			return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 58, Col: 71}
+			return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 58, Col: 71}
 		}
 		_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var3))
 		if templ_7745c5c3_Err != nil {
@@ -67,7 +67,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 		var templ_7745c5c3_Var4 string
 		templ_7745c5c3_Var4, templ_7745c5c3_Err = templ.JoinStringErrs(data.LastUpdated.Format("15:04"))
 		if templ_7745c5c3_Err != nil {
-			return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 78, Col: 69}
+			return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 78, Col: 69}
 		}
 		_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var4))
 		if templ_7745c5c3_Err != nil {
@@ -85,7 +85,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 			var templ_7745c5c3_Var5 string
 			templ_7745c5c3_Var5, templ_7745c5c3_Err = templ.JoinStringErrs(user.Username)
 			if templ_7745c5c3_Err != nil {
-				return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 127, Col: 74}
+				return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 127, Col: 74}
 			}
 			_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var5))
 			if templ_7745c5c3_Err != nil {
@@ -108,7 +108,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 			var templ_7745c5c3_Var6 string
 			templ_7745c5c3_Var6, templ_7745c5c3_Err = templ.JoinStringErrs(user.Email)
 			if templ_7745c5c3_Err != nil {
-				return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 133, Col: 59}
+				return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 133, Col: 59}
 			}
 			_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var6))
 			if templ_7745c5c3_Err != nil {
@@ -121,7 +121,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 			var templ_7745c5c3_Var7 string
 			templ_7745c5c3_Var7, templ_7745c5c3_Err = templ.JoinStringErrs(user.AccessKey)
 			if templ_7745c5c3_Err != nil {
-				return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 135, Col: 88}
+				return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 135, Col: 88}
 			}
 			_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var7))
 			if templ_7745c5c3_Err != nil {
@@ -134,7 +134,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 			var templ_7745c5c3_Var8 string
 			templ_7745c5c3_Var8, templ_7745c5c3_Err = templ.JoinStringErrs(user.Username)
 			if templ_7745c5c3_Err != nil {
-				return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 140, Col: 121}
+				return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 140, Col: 121}
 			}
 			_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var8))
 			if templ_7745c5c3_Err != nil {
@@ -152,7 +152,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 				var templ_7745c5c3_Var9 string
 				templ_7745c5c3_Var9, templ_7745c5c3_Err = templ.JoinStringErrs(user.Username)
 				if templ_7745c5c3_Err != nil {
-					return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 145, Col: 117}
+					return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 145, Col: 117}
 				}
 				_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var9))
 				if templ_7745c5c3_Err != nil {
@@ -171,7 +171,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 				var templ_7745c5c3_Var10 string
 				templ_7745c5c3_Var10, templ_7745c5c3_Err = templ.JoinStringErrs(user.Username)
 				if templ_7745c5c3_Err != nil {
-					return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 151, Col: 126}
+					return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 151, Col: 126}
 				}
 				_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var10))
 				if templ_7745c5c3_Err != nil {
@@ -190,7 +190,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 				var templ_7745c5c3_Var11 string
 				templ_7745c5c3_Var11, templ_7745c5c3_Err = templ.JoinStringErrs(user.Username)
 				if templ_7745c5c3_Err != nil {
-					return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 157, Col: 119}
+					return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 157, Col: 119}
 				}
 				_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var11))
 				if templ_7745c5c3_Err != nil {
@@ -219,7 +219,7 @@ func ObjectStoreUsers(data dash.ObjectStoreUsersData) templ.Component {
 		var templ_7745c5c3_Var12 string
 		templ_7745c5c3_Var12, templ_7745c5c3_Err = templ.JoinStringErrs(data.LastUpdated.Format("2006-01-02 15:04:05"))
 		if templ_7745c5c3_Err != nil {
-			return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/object_store_users.templ`, Line: 189, Col: 81}
+			return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/object_store_users.templ`, Line: 189, Col: 81}
 		}
 		_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var12))
 		if templ_7745c5c3_Err != nil {

weed/command/filer_backup.go

@@ -11,6 +11,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/repl_util"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -143,6 +144,10 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 		sourceFiler.ToGrpcAddress(),
 		sourcePath,
 		*backupOption.proxyByFiler)
+
+	if err := repl_util.InitializeSSEForReplication(filerSource); err != nil {
+		return fmt.Errorf("SSE initialization failed: %v", err)
+	}
 	dataSink.SetSourceFiler(filerSource)
 
 	var processEventFn func(*filer_pb.SubscribeMetadataResponse) error

weed/command/master.go

@@ -96,7 +96,7 @@ func init() {
 	m.metricsIntervalSec = cmdMaster.Flag.Int("metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
 	m.metricsHttpPort = cmdMaster.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 	m.metricsHttpIp = cmdMaster.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
-	m.raftResumeState = cmdMaster.Flag.Bool("resumeState", false, "resume previous state on start master server")
+	m.raftResumeState = cmdMaster.Flag.Bool("resumeState", true, "resume previous state on start master server")
 	m.heartbeatInterval = cmdMaster.Flag.Duration("heartbeatInterval", 300*time.Millisecond, "heartbeat interval of master servers, and will be randomly multiplied by [1, 1.25)")
 	m.electionTimeout = cmdMaster.Flag.Duration("electionTimeout", 10*time.Second, "election timeout of master servers")
 	m.raftHashicorp = cmdMaster.Flag.Bool("raftHashicorp", false, "use hashicorp raft")
@@ -208,6 +208,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 		DataDir:           util.ResolvePath(metaDir),
 		Topo:              ms.Topo,
 		RaftResumeState:   *masterOption.raftResumeState,
+		SingleMaster:      isSingleMaster,
 		HeartbeatInterval: *masterOption.heartbeatInterval,
 		ElectionTimeout:   *masterOption.electionTimeout,
 		RaftBootstrap:     *masterOption.raftBootstrap,

weed/command/mini.go

@@ -162,7 +162,7 @@ func initMiniMasterFlags() {
 	miniMasterOptions.garbageThreshold = cmdMini.Flag.Float64("master.garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces")
 	miniMasterOptions.metricsAddress = cmdMini.Flag.String("master.metrics.address", "", "Prometheus gateway address")
 	miniMasterOptions.metricsIntervalSec = cmdMini.Flag.Int("master.metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
-	miniMasterOptions.raftResumeState = cmdMini.Flag.Bool("master.resumeState", false, "resume previous state on start master server")
+	miniMasterOptions.raftResumeState = cmdMini.Flag.Bool("master.resumeState", true, "resume previous state on start master server")
 	miniMasterOptions.heartbeatInterval = cmdMini.Flag.Duration("master.heartbeatInterval", 300*time.Millisecond, "heartbeat interval of master servers, and will be randomly multiplied by [1, 1.25)")
 	miniMasterOptions.electionTimeout = cmdMini.Flag.Duration("master.electionTimeout", 10*time.Second, "election timeout of master servers")
 	miniMasterOptions.raftHashicorp = cmdMini.Flag.Bool("master.raftHashicorp", false, "use hashicorp raft")

weed/command/server.go

@@ -102,7 +102,7 @@ func init() {
 	masterOptions.garbageThreshold = cmdServer.Flag.Float64("master.garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces")
 	masterOptions.metricsAddress = cmdServer.Flag.String("master.metrics.address", "", "Prometheus gateway address")
 	masterOptions.metricsIntervalSec = cmdServer.Flag.Int("master.metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
-	masterOptions.raftResumeState = cmdServer.Flag.Bool("master.resumeState", false, "resume previous state on start master server")
+	masterOptions.raftResumeState = cmdServer.Flag.Bool("master.resumeState", true, "resume previous state on start master server")
 	masterOptions.raftHashicorp = cmdServer.Flag.Bool("master.raftHashicorp", false, "use hashicorp raft")
 	masterOptions.raftBootstrap = cmdServer.Flag.Bool("master.raftBootstrap", false, "Whether to bootstrap the Raft cluster")
 	masterOptions.heartbeatInterval = cmdServer.Flag.Duration("master.heartbeatInterval", 300*time.Millisecond, "heartbeat interval of master servers, and will be randomly multiplied by [1, 1.25)")

weed/iam/policy/aws_iam_compliance_test.go

@@ -110,6 +110,92 @@ func TestAWSIAMMatch(t *testing.T) {
 	}
 }
 
+func TestMatchesActionsMultipartExpansion(t *testing.T) {
+	engine := &PolicyEngine{initialized: true}
+	evalCtx := &EvaluationContext{}
+
+	tests := []struct {
+		name            string
+		actions         []string
+		requestedAction string
+		expected        bool
+	}{
+		{
+			name:            "PutObject directly matches PutObject",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:PutObject",
+			expected:        true,
+		},
+		{
+			name:            "PutObject implicitly allows CreateMultipartUpload",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:CreateMultipartUpload",
+			expected:        true,
+		},
+		{
+			name:            "PutObject implicitly allows UploadPart",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:UploadPart",
+			expected:        true,
+		},
+		{
+			name:            "PutObject implicitly allows CompleteMultipartUpload",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:CompleteMultipartUpload",
+			expected:        true,
+		},
+		{
+			name:            "PutObject implicitly allows AbortMultipartUpload",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:AbortMultipartUpload",
+			expected:        true,
+		},
+		{
+			name:            "PutObject implicitly allows ListMultipartUploadParts",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:ListMultipartUploadParts",
+			expected:        true,
+		},
+		{
+			name:            "PutObject implicitly allows ListBucketMultipartUploads",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:ListBucketMultipartUploads",
+			expected:        true,
+		},
+		{
+			name:            "PutObject does not allow GetObject",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "s3:GetObject",
+			expected:        false,
+		},
+		{
+			name:            "GetObject does not allow CreateMultipartUpload",
+			actions:         []string{"s3:GetObject"},
+			requestedAction: "s3:CreateMultipartUpload",
+			expected:        false,
+		},
+		{
+			name:            "wildcard s3:Put* implicitly allows multipart via PutObject match",
+			actions:         []string{"s3:Put*"},
+			requestedAction: "s3:CreateMultipartUpload",
+			expected:        true,
+		},
+		{
+			name:            "case-insensitive multipart action lookup",
+			actions:         []string{"s3:PutObject"},
+			requestedAction: "S3:CREATEMULTIPARTUPLOAD",
+			expected:        true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := engine.matchesActions(tt.actions, tt.requestedAction, evalCtx)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
 func TestExpandPolicyVariables(t *testing.T) {
 	evalCtx := &EvaluationContext{
 		RequestContext: map[string]interface{}{

weed/iam/policy/policy_engine.go

@@ -600,12 +600,31 @@ func (e *PolicyEngine) statementMatches(statement *Statement, evalCtx *Evaluatio
 	return true
 }
 
-// matchesActions checks if any action in the list matches the requested action
+// multipartActionSet contains lowercased S3 multipart upload actions that are
+// implicitly granted when s3:PutObject is allowed, since multipart upload is an
+// implementation detail of putting objects. Keys are lowercased for
+// case-insensitive lookup (AWS IAM actions are case-insensitive).
+var multipartActionSet = map[string]bool{
+	"s3:createmultipartupload":      true,
+	"s3:uploadpart":                 true,
+	"s3:completemultipartupload":    true,
+	"s3:abortmultipartupload":       true,
+	"s3:listmultipartuploadparts":   true,
+	"s3:listbucketmultipartuploads": true,
+}
+
+// matchesActions checks if any action in the list matches the requested action.
+// It also implicitly grants multipart upload actions when s3:PutObject is allowed,
+// mirroring the behavior in the S3 API policy engine (see PR #8445).
 func (e *PolicyEngine) matchesActions(actions []string, requestedAction string, evalCtx *EvaluationContext) bool {
+	isMultipart := multipartActionSet[strings.ToLower(requestedAction)]
 	for _, action := range actions {
 		if awsIAMMatch(action, requestedAction, evalCtx) {
 			return true
 		}
+		if isMultipart && awsIAMMatch(action, "s3:PutObject", evalCtx) {
+			return true
+		}
 	}
 	return false
 }

weed/mq/kafka/consumer/rebalance_timeout.go

@@ -116,11 +116,9 @@ func (rtm *RebalanceTimeoutManager) IsRebalanceStuck(group *ConsumerGroup, maxRe
 	return time.Since(group.LastActivity) > maxRebalanceDuration
 }
 
-// ForceCompleteRebalance forces completion of a stuck rebalance
+// ForceCompleteRebalance forces completion of a stuck rebalance.
+// IMPORTANT: The caller must already hold group.Mu.Lock().
 func (rtm *RebalanceTimeoutManager) ForceCompleteRebalance(group *ConsumerGroup) {
-	group.Mu.Lock()
-	defer group.Mu.Unlock()
-
 	// If stuck in preparing rebalance, move to completing
 	if group.State == GroupStatePreparingRebalance {
 		group.State = GroupStateCompletingRebalance

weed/mq/kafka/consumer/rebalance_timeout_test.go

@@ -185,20 +185,19 @@ func TestRebalanceTimeoutManager_ForceCompleteRebalance(t *testing.T) {
 		State: MemberStatePending,
 	}
 	group.Members["member1"] = member
-	group.Mu.Unlock()
 
+	// ForceCompleteRebalance expects the caller to hold group.Mu.Lock()
 	rtm.ForceCompleteRebalance(group)
 
-	group.Mu.RLock()
 	if group.State != GroupStateCompletingRebalance {
 		t.Errorf("Expected group state to be CompletingRebalance, got %s", group.State.String())
 	}
-	group.Mu.RUnlock()
+	group.Mu.Unlock()
 
 	// Test forcing completion from CompletingRebalance
+	group.Mu.Lock()
 	rtm.ForceCompleteRebalance(group)
 
-	group.Mu.RLock()
 	if group.State != GroupStateStable {
 		t.Errorf("Expected group state to be Stable, got %s", group.State.String())
 	}
@@ -206,7 +205,7 @@ func TestRebalanceTimeoutManager_ForceCompleteRebalance(t *testing.T) {
 	if member.State != MemberStateStable {
 		t.Errorf("Expected member state to be Stable, got %s", member.State.String())
 	}
-	group.Mu.RUnlock()
+	group.Mu.Unlock()
 }
 
 func TestRebalanceTimeoutManager_GetRebalanceStatus(t *testing.T) {

weed/mq/kafka/protocol/fetch.go

@@ -55,7 +55,10 @@ func (h *Handler) handleFetch(ctx context.Context, correlationID uint32, apiVers
 			for _, partition := range topic.Partitions {
 				hwm, err := h.seaweedMQHandler.GetLatestOffset(topic.Name, partition.PartitionID)
 				if err != nil {
-					continue
+					// HWM lookup failed (e.g. partition deactivated between consumer
+					// sessions). Assume data may be available rather than blocking in
+					// the long-poll loop — the actual fetch will determine the truth.
+					return true
 				}
 				// Normalize fetch offset
 				effectiveOffset := partition.FetchOffset

weed/mq/kafka/protocol/fetch_partition_reader.go

@@ -2,6 +2,7 @@ package protocol
 
 import (
 	"context"
+	"math"
 	"sync"
 	"time"
 
@@ -136,13 +137,19 @@ func (pr *partitionReader) serveFetchRequest(ctx context.Context, req *partition
 	}()
 
 	// Get high water mark
+	hwmUnknown := false
 	hwm, hwmErr := pr.handler.seaweedMQHandler.GetLatestOffset(pr.topicName, pr.partitionID)
 	if hwmErr != nil {
-		glog.Errorf("[%s] CRITICAL: Failed to get HWM for %s[%d]: %v",
+		// HWM lookup can fail when the partition has been deactivated between consumer
+		// sessions. Proceed with the fetch anyway — the broker will return the correct
+		// data (or empty) based on its own state. Use math.MaxInt64 as sentinel so
+		// FetchMultipleBatches doesn't artificially cap recordsAvailable, and we
+		// don't hit the early-return below. The actual HWM will be derived from
+		// the fetch result (newOffset) after the read.
+		glog.Warningf("[%s] HWM lookup failed for %s[%d]: %v — will attempt fetch anyway",
 			pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, hwmErr)
-		result.recordBatch = []byte{}
-		result.highWaterMark = 0
-		return
+		hwm = math.MaxInt64
+		hwmUnknown = true
 	}
 	result.highWaterMark = hwm
 
@@ -170,10 +177,20 @@ func (pr *partitionReader) serveFetchRequest(ctx context.Context, req *partition
 	// Fetch on-demand - no pre-fetching to avoid overwhelming the broker
 	recordBatch, newOffset := pr.readRecords(ctx, req.requestedOffset, req.maxBytes, req.maxWaitMs, hwm)
 
+	// When HWM was unknown, derive a reasonable value from the fetch result
+	// so the client sees a meaningful high water mark instead of MaxInt64.
+	if hwmUnknown {
+		if newOffset > req.requestedOffset {
+			result.highWaterMark = newOffset // best estimate: end of what we read
+		} else {
+			result.highWaterMark = req.requestedOffset // no data found
+		}
+	}
+
 	// Log what we got back - DETAILED for diagnostics
 	if len(recordBatch) == 0 {
 		glog.V(2).Infof("[%s] FETCH %s[%d]: readRecords returned EMPTY (offset=%d, hwm=%d)",
-			pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, req.requestedOffset, hwm)
+			pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, req.requestedOffset, result.highWaterMark)
 		result.recordBatch = []byte{}
 	} else {
 		result.recordBatch = recordBatch
@@ -228,15 +245,40 @@ func (pr *partitionReader) readRecords(ctx context.Context, fromOffset int64, ma
 		return fetchResult.RecordBatches, fetchResult.NextOffset
 	}
 
-	// Multi-batch failed - try single batch WITHOUT the timeout constraint
-	// to ensure we get at least some data even if multi-batch timed out
+	// Multi-batch failed - try single batch with a fresh timeout
 	glog.Warningf("[%s] Multi-batch fetch failed for %s[%d] offset=%d after %v, falling back to single-batch (err: %v)",
 		pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, fromOffset, fetchDuration, err)
 
-	// Use original context for fallback, NOT the timed-out fetchCtx
-	// This ensures the fallback has a fresh chance to fetch data
+	// Compute the remaining time budget for the fallback. If the parent
+	// context carries a deadline, honour it; otherwise derive remaining
+	// from maxWaitMs minus elapsed time. This prevents the fallback from
+	// restarting the full budget after the multi-batch fetch already
+	// consumed part of it.
+	var remaining time.Duration
+	if deadline, ok := ctx.Deadline(); ok {
+		remaining = time.Until(deadline)
+	} else {
+		remaining = time.Duration(maxWaitMs)*time.Millisecond - time.Since(fetchStartTime)
+	}
+	if remaining <= 0 {
+		// Budget exhausted — skip the fallback entirely.
+		glog.V(2).Infof("[%s] No remaining budget for fallback on %s[%d] (maxWait=%dms, elapsed=%v)",
+			pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, maxWaitMs, time.Since(fetchStartTime))
+		return []byte{}, fromOffset
+	}
+	// Clamp: floor of 2s so disk reads via gRPC have a realistic chance,
+	// but never exceed 10s to bound data-plane blocking.
+	fallbackTimeout := remaining
+	if fallbackTimeout < 2*time.Second {
+		fallbackTimeout = 2 * time.Second
+	}
+	if fallbackTimeout > 10*time.Second {
+		fallbackTimeout = 10 * time.Second
+	}
+	fallbackCtx, fallbackCancel := context.WithTimeout(ctx, fallbackTimeout)
+	defer fallbackCancel()
 	fallbackStartTime := time.Now()
-	smqRecords, err := pr.handler.seaweedMQHandler.GetStoredRecords(ctx, pr.topicName, pr.partitionID, fromOffset, 10)
+	smqRecords, err := pr.handler.seaweedMQHandler.GetStoredRecords(fallbackCtx, pr.topicName, pr.partitionID, fromOffset, 10)
 	fallbackDuration := time.Since(fallbackStartTime)
 
 	if fallbackDuration > 2*time.Second {

weed/replication/repl_util/replication_util.go

@@ -2,14 +2,73 @@ package repl_util
 
 import (
 	"context"
+	"io"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
-func CopyFromChunkViews(chunkViews *filer.IntervalList[*filer.ChunkView], filerSource *source.FilerSource, writeFunc func(data []byte) error) error {
+// CopyFromChunkViews copies chunk data with optional SSE decryption.
+// If entry has SSE-encrypted chunks, data is decrypted before writing.
+func CopyFromChunkViews(chunkViews *filer.IntervalList[*filer.ChunkView], filerSource *source.FilerSource, writeFunc func(data []byte) error, entry *filer_pb.Entry) error {
+	if entry != nil {
+		sseType, err := detectSSEType(entry)
+		if err != nil {
+			return err
+		}
+		if sseType != filer_pb.SSEType_NONE {
+			return copyWithDecryption(filerSource, entry, writeFunc)
+		}
+	}
+	return copyChunkViews(chunkViews, filerSource, writeFunc)
+}
+
+func copyWithDecryption(filerSource *source.FilerSource, entry *filer_pb.Entry, writeFunc func(data []byte) error) error {
+	reader := filer.NewFileReader(filerSource, entry)
+	decrypted, err := MaybeDecryptReader(reader, entry)
+	if err != nil {
+		CloseReader(reader)
+		return err
+	}
+	defer CloseMaybeDecryptedReader(reader, decrypted)
+	buf := make([]byte, 128*1024)
+	for {
+		n, readErr := decrypted.Read(buf)
+		if n > 0 {
+			if writeErr := writeFunc(buf[:n]); writeErr != nil {
+				return writeErr
+			}
+		}
+		if readErr == io.EOF {
+			return nil
+		}
+		if readErr != nil {
+			return readErr
+		}
+	}
+}
+
+// CloseReader closes r if it implements io.Closer.
+func CloseReader(r io.Reader) {
+	if closer, ok := r.(io.Closer); ok {
+		closer.Close()
+	}
+}
+
+// CloseMaybeDecryptedReader closes the decrypted reader if it implements io.Closer,
+// otherwise falls back to closing the original reader.
+func CloseMaybeDecryptedReader(original, decrypted io.Reader) {
+	if closer, ok := decrypted.(io.Closer); ok {
+		closer.Close()
+	} else {
+		CloseReader(original)
+	}
+}
+
+func copyChunkViews(chunkViews *filer.IntervalList[*filer.ChunkView], filerSource *source.FilerSource, writeFunc func(data []byte) error) error {
 
 	for x := chunkViews.Front(); x != nil; x = x.Next {
 		chunk := x.Value

weed/replication/repl_util/sse_init.go

@@ -0,0 +1,55 @@
+package repl_util
+
+import (
+	"sync"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/kms"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+var (
+	sseInitMu          sync.Mutex
+	sseInitialized     bool
+)
+
+// InitializeSSEForReplication sets up SSE-S3 and SSE-KMS decryption so that
+// replication sinks can transparently decrypt encrypted objects.
+// SSE-S3 is initialized from the filer (KEK stored on filer).
+// SSE-KMS is initialized from Viper config (security.toml [kms] section or
+// WEED_KMS_* environment variables).
+// SSE-C cannot be decrypted (customer key not available) and will error at
+// decryption time.
+//
+// Safe to call multiple times; only the first successful initialization takes
+// effect. Failed attempts do not prevent future retries.
+func InitializeSSEForReplication(filerSource filer_pb.FilerClient) error {
+	sseInitMu.Lock()
+	defer sseInitMu.Unlock()
+	if sseInitialized {
+		return nil
+	}
+
+	// Initialize SSE-S3 key manager from filer
+	if err := s3api.GetSSES3KeyManager().InitializeWithFiler(filerSource); err != nil {
+		return err
+	}
+
+	// Attempt KMS initialization from Viper config.
+	// KMS configuration is typically in the S3 config file which the
+	// replication commands don't load directly. Support loading from
+	// security.toml [kms] section or WEED_KMS_* environment variables.
+	loader := kms.NewConfigLoader(util.GetViper())
+	if err := loader.LoadConfigurations(); err != nil {
+		glog.Warningf("KMS initialization from config failed: %v (SSE-KMS decryption will not be available)", err)
+	} else if err := loader.ValidateConfiguration(); err != nil {
+		glog.Warningf("KMS configuration validation failed: %v (SSE-KMS decryption will not be available)", err)
+	} else {
+		glog.V(0).Infof("KMS initialized for replication")
+	}
+
+	sseInitialized = true
+	return nil
+}

weed/replication/repl_util/sse_reader.go

@@ -0,0 +1,150 @@
+package repl_util
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+// MaybeDecryptReader wraps reader with SSE decryption if the entry has encrypted chunks.
+// Returns the original reader unchanged if no SSE encryption is detected.
+func MaybeDecryptReader(reader io.Reader, entry *filer_pb.Entry) (io.Reader, error) {
+	if entry == nil {
+		return reader, nil
+	}
+
+	sseType, err := detectSSEType(entry)
+	if err != nil {
+		return nil, err
+	}
+	if sseType == filer_pb.SSEType_NONE {
+		return reader, nil
+	}
+
+	switch sseType {
+	case filer_pb.SSEType_SSE_S3:
+		return decryptSSES3(reader, entry)
+	case filer_pb.SSEType_SSE_KMS:
+		return decryptSSEKMS(reader, entry)
+	case filer_pb.SSEType_SSE_C:
+		return nil, fmt.Errorf("SSE-C encrypted object cannot be decrypted during replication (customer key not available)")
+	}
+
+	return nil, fmt.Errorf("unsupported SSE type: %v", sseType)
+}
+
+// MaybeDecryptContent decrypts inline entry content if SSE-encrypted.
+// Returns the original content unchanged if no SSE encryption is detected.
+func MaybeDecryptContent(content []byte, entry *filer_pb.Entry) ([]byte, error) {
+	if entry == nil || len(content) == 0 {
+		return content, nil
+	}
+
+	sseType, err := detectSSEType(entry)
+	if err != nil {
+		return nil, err
+	}
+	if sseType == filer_pb.SSEType_NONE {
+		return content, nil
+	}
+
+	reader := bytes.NewReader(content)
+	decrypted, err := MaybeDecryptReader(reader, entry)
+	if err != nil {
+		return nil, err
+	}
+	return io.ReadAll(decrypted)
+}
+
+func detectSSEType(entry *filer_pb.Entry) (filer_pb.SSEType, error) {
+	// Check chunk metadata first
+	var detected filer_pb.SSEType
+	for _, chunk := range entry.GetChunks() {
+		if chunk.SseType != filer_pb.SSEType_NONE {
+			if detected == filer_pb.SSEType_NONE {
+				detected = chunk.SseType
+			} else if chunk.SseType != detected {
+				return filer_pb.SSEType_NONE, fmt.Errorf("mixed SSE types in chunks: %v and %v", detected, chunk.SseType)
+			}
+		}
+	}
+	if detected != filer_pb.SSEType_NONE {
+		return detected, nil
+	}
+
+	// Fall back to extended metadata for inline objects (no chunks)
+	if entry.Extended != nil {
+		hasS3 := len(entry.Extended[s3_constants.SeaweedFSSSES3Key]) > 0
+		hasKMS := len(entry.Extended[s3_constants.SeaweedFSSSEKMSKey]) > 0
+		hasC := len(entry.Extended[s3_constants.SeaweedFSSSEIV]) > 0
+		count := 0
+		if hasS3 {
+			count++
+		}
+		if hasKMS {
+			count++
+		}
+		if hasC {
+			count++
+		}
+		if count > 1 {
+			return filer_pb.SSEType_NONE, fmt.Errorf("conflicting SSE metadata in entry: multiple SSE key types present")
+		}
+		if hasS3 {
+			return filer_pb.SSEType_SSE_S3, nil
+		}
+		if hasKMS {
+			return filer_pb.SSEType_SSE_KMS, nil
+		}
+		if hasC {
+			return filer_pb.SSEType_SSE_C, nil
+		}
+	}
+	return filer_pb.SSEType_NONE, nil
+}
+
+func decryptSSES3(reader io.Reader, entry *filer_pb.Entry) (io.Reader, error) {
+	if entry.Extended == nil {
+		return nil, fmt.Errorf("SSE-S3 encrypted entry has no extended metadata")
+	}
+
+	keyData := entry.Extended[s3_constants.SeaweedFSSSES3Key]
+	if len(keyData) == 0 {
+		return nil, fmt.Errorf("SSE-S3 key metadata not found in entry")
+	}
+
+	keyManager := s3api.GetSSES3KeyManager()
+	sseS3Key, err := s3api.DeserializeSSES3Metadata(keyData, keyManager)
+	if err != nil {
+		return nil, fmt.Errorf("deserialize SSE-S3 metadata: %w", err)
+	}
+
+	iv, err := s3api.GetSSES3IV(entry, sseS3Key, keyManager)
+	if err != nil {
+		return nil, fmt.Errorf("get SSE-S3 IV: %w", err)
+	}
+
+	return s3api.CreateSSES3DecryptedReader(reader, sseS3Key, iv)
+}
+
+func decryptSSEKMS(reader io.Reader, entry *filer_pb.Entry) (io.Reader, error) {
+	if entry.Extended == nil {
+		return nil, fmt.Errorf("SSE-KMS encrypted entry has no extended metadata")
+	}
+
+	kmsMetadata := entry.Extended[s3_constants.SeaweedFSSSEKMSKey]
+	if len(kmsMetadata) == 0 {
+		return nil, fmt.Errorf("SSE-KMS key metadata not found in entry")
+	}
+
+	sseKMSKey, err := s3api.DeserializeSSEKMSMetadata(kmsMetadata)
+	if err != nil {
+		return nil, fmt.Errorf("deserialize SSE-KMS metadata: %w", err)
+	}
+
+	return s3api.CreateSSEKMSDecryptedReader(reader, sseKMSKey)
+}

weed/replication/repl_util/sse_reader_test.go

@@ -0,0 +1,534 @@
+package repl_util
+
+import (
+	"bytes"
+	"context"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/kms"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func TestDetectSSEType(t *testing.T) {
+	tests := []struct {
+		name      string
+		entry     *filer_pb.Entry
+		wantType  filer_pb.SSEType
+		wantError bool
+	}{
+		{
+			name:     "no chunks no extended",
+			entry:    &filer_pb.Entry{},
+			wantType: filer_pb.SSEType_NONE,
+		},
+		{
+			name: "plaintext chunks",
+			entry: &filer_pb.Entry{
+				Chunks: []*filer_pb.FileChunk{
+					{SseType: filer_pb.SSEType_NONE},
+					{SseType: filer_pb.SSEType_NONE},
+				},
+			},
+			wantType: filer_pb.SSEType_NONE,
+		},
+		{
+			name: "uniform SSE-S3 chunks",
+			entry: &filer_pb.Entry{
+				Chunks: []*filer_pb.FileChunk{
+					{SseType: filer_pb.SSEType_SSE_S3},
+					{SseType: filer_pb.SSEType_SSE_S3},
+				},
+			},
+			wantType: filer_pb.SSEType_SSE_S3,
+		},
+		{
+			name: "uniform SSE-KMS chunks",
+			entry: &filer_pb.Entry{
+				Chunks: []*filer_pb.FileChunk{
+					{SseType: filer_pb.SSEType_SSE_KMS},
+				},
+			},
+			wantType: filer_pb.SSEType_SSE_KMS,
+		},
+		{
+			name: "mixed chunk SSE types",
+			entry: &filer_pb.Entry{
+				Chunks: []*filer_pb.FileChunk{
+					{SseType: filer_pb.SSEType_SSE_S3},
+					{SseType: filer_pb.SSEType_SSE_KMS},
+				},
+			},
+			wantError: true,
+		},
+		{
+			name: "inline SSE-S3 via extended",
+			entry: &filer_pb.Entry{
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSSSES3Key: {0x01},
+				},
+			},
+			wantType: filer_pb.SSEType_SSE_S3,
+		},
+		{
+			name: "inline SSE-KMS via extended",
+			entry: &filer_pb.Entry{
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSSSEKMSKey: {0x01},
+				},
+			},
+			wantType: filer_pb.SSEType_SSE_KMS,
+		},
+		{
+			name: "inline SSE-C via extended",
+			entry: &filer_pb.Entry{
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSSSEIV: {0x01},
+				},
+			},
+			wantType: filer_pb.SSEType_SSE_C,
+		},
+		{
+			name: "conflicting extended metadata",
+			entry: &filer_pb.Entry{
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSSSES3Key:  {0x01},
+					s3_constants.SeaweedFSSSEKMSKey: {0x02},
+				},
+			},
+			wantError: true,
+		},
+		{
+			name: "chunks take precedence over extended",
+			entry: &filer_pb.Entry{
+				Chunks: []*filer_pb.FileChunk{
+					{SseType: filer_pb.SSEType_SSE_S3},
+				},
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSSSEKMSKey: {0x01},
+				},
+			},
+			wantType: filer_pb.SSEType_SSE_S3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := detectSSEType(tt.entry)
+			if tt.wantError {
+				if err == nil {
+					t.Fatalf("expected error, got type %v", got)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if got != tt.wantType {
+				t.Errorf("got %v, want %v", got, tt.wantType)
+			}
+		})
+	}
+}
+
+func TestMaybeDecryptReader_Plaintext(t *testing.T) {
+	content := []byte("hello world")
+	entry := &filer_pb.Entry{}
+	reader := bytes.NewReader(content)
+
+	got, err := MaybeDecryptReader(reader, entry)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	result, err := io.ReadAll(got)
+	if err != nil {
+		t.Fatalf("ReadAll error: %v", err)
+	}
+	if !bytes.Equal(result, content) {
+		t.Errorf("got %q, want %q", result, content)
+	}
+}
+
+func TestMaybeDecryptReader_NilEntry(t *testing.T) {
+	content := []byte("hello")
+	reader := bytes.NewReader(content)
+
+	got, err := MaybeDecryptReader(reader, nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	result, err := io.ReadAll(got)
+	if err != nil {
+		t.Fatalf("ReadAll error: %v", err)
+	}
+	if !bytes.Equal(result, content) {
+		t.Errorf("got %q, want %q", result, content)
+	}
+}
+
+func TestMaybeDecryptReader_SSEC_Error(t *testing.T) {
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSEIV: {0x01},
+		},
+	}
+	reader := bytes.NewReader([]byte("data"))
+
+	_, err := MaybeDecryptReader(reader, entry)
+	if err == nil {
+		t.Fatal("expected error for SSE-C")
+	}
+}
+
+func TestMaybeDecryptContent_Plaintext(t *testing.T) {
+	content := []byte("hello world")
+	entry := &filer_pb.Entry{}
+
+	got, err := MaybeDecryptContent(content, entry)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !bytes.Equal(got, content) {
+		t.Errorf("got %q, want %q", got, content)
+	}
+}
+
+func TestMaybeDecryptContent_NilEntry(t *testing.T) {
+	content := []byte("data")
+	got, err := MaybeDecryptContent(content, nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !bytes.Equal(got, content) {
+		t.Errorf("got %q, want %q", got, content)
+	}
+}
+
+func TestMaybeDecryptContent_Empty(t *testing.T) {
+	got, err := MaybeDecryptContent(nil, &filer_pb.Entry{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got != nil {
+		t.Errorf("expected nil, got %v", got)
+	}
+}
+
+func TestMaybeDecryptContent_SSEC_Error(t *testing.T) {
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSEIV: {0x01},
+		},
+	}
+
+	_, err := MaybeDecryptContent([]byte("data"), entry)
+	if err == nil {
+		t.Fatal("expected error for SSE-C")
+	}
+}
+
+func TestMaybeDecryptContent_MixedExtended_Error(t *testing.T) {
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSES3Key:  {0x01},
+			s3_constants.SeaweedFSSSEKMSKey: {0x02},
+		},
+	}
+
+	_, err := MaybeDecryptContent([]byte("data"), entry)
+	if err == nil {
+		t.Fatal("expected error for conflicting SSE metadata")
+	}
+}
+
+// --- SSE-S3 integration tests ---
+// These tests exercise the full MaybeDecryptReader/MaybeDecryptContent path
+// for SSE-S3: detectSSEType → decryptSSES3 → DeserializeSSES3Metadata →
+// GetSSES3IV → CreateSSES3DecryptedReader. A test KEK is injected via
+// WEED_S3_SSE_KEK env var and a mock filer client.
+
+// testFilerClient is a minimal filer_pb.FilerClient mock that returns
+// ErrNotFound for all lookups (no KEK on filer — we use env var instead).
+type testFilerClient struct{}
+
+func (c *testFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fmt.Errorf("%w", filer_pb.ErrNotFound)
+}
+func (c *testFilerClient) AdjustedUrl(loc *filer_pb.Location) string { return loc.Url }
+func (c *testFilerClient) GetDataCenter() string                     { return "" }
+
+// setupTestSSES3 initializes the global SSE-S3 key manager with a test KEK
+// via the WEED_S3_SSE_KEK env var and returns the KEK bytes + cleanup func.
+func setupTestSSES3(t *testing.T) (kek []byte, cleanup func()) {
+	t.Helper()
+
+	kek = make([]byte, 32)
+	if _, err := io.ReadFull(rand.Reader, kek); err != nil {
+		t.Fatal(err)
+	}
+
+	// Force Viper to pick up the new env var
+	os.Setenv("WEED_S3_SSE_KEK", hex.EncodeToString(kek))
+
+	// Reset Viper cache so it reads the new env var
+	v := util.GetViper()
+	v.AutomaticEnv()
+
+	// Re-initialize the global key manager with the KEK from env
+	km := s3api.GetSSES3KeyManager()
+	if err := km.InitializeWithFiler(&testFilerClient{}); err != nil {
+		os.Unsetenv("WEED_S3_SSE_KEK")
+		t.Fatalf("InitializeWithFiler: %v", err)
+	}
+
+	return kek, func() {
+		os.Unsetenv("WEED_S3_SSE_KEK")
+		// Re-initialize with no KEK to clear the super key
+		km.InitializeWithFiler(&testFilerClient{})
+	}
+}
+
+func TestMaybeDecryptReader_SSES3(t *testing.T) {
+	_, cleanup := setupTestSSES3(t)
+	defer cleanup()
+
+	plaintext := []byte("SSE-S3 encrypted content for testing round-trip decryption")
+
+	// Generate a DEK and encrypt
+	sseKey, err := s3api.GenerateSSES3Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	encReader, encIV, err := s3api.CreateSSES3EncryptedReader(bytes.NewReader(plaintext), sseKey)
+	if err != nil {
+		t.Fatalf("encrypt: %v", err)
+	}
+	ciphertext, err := io.ReadAll(encReader)
+	if err != nil {
+		t.Fatalf("read ciphertext: %v", err)
+	}
+
+	// Build serialized SSE-S3 metadata (uses the global key manager to
+	// envelope-encrypt the DEK with the test KEK)
+	sseKey.IV = encIV
+	metadataBytes, err := s3api.SerializeSSES3Metadata(sseKey)
+	if err != nil {
+		t.Fatalf("serialize metadata: %v", err)
+	}
+
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSES3Key: metadataBytes,
+		},
+	}
+
+	// Test full path: MaybeDecryptReader → decryptSSES3 → DeserializeSSES3Metadata → CreateSSES3DecryptedReader
+	decrypted, err := MaybeDecryptReader(bytes.NewReader(ciphertext), entry)
+	if err != nil {
+		t.Fatalf("MaybeDecryptReader: %v", err)
+	}
+	result, err := io.ReadAll(decrypted)
+	if err != nil {
+		t.Fatalf("ReadAll: %v", err)
+	}
+	if !bytes.Equal(result, plaintext) {
+		t.Errorf("SSE-S3 round-trip failed: got %q, want %q", result, plaintext)
+	}
+}
+
+func TestMaybeDecryptContent_SSES3(t *testing.T) {
+	_, cleanup := setupTestSSES3(t)
+	defer cleanup()
+
+	plaintext := []byte("inline SSE-S3 content")
+
+	// Generate a DEK and encrypt inline content
+	sseKey, err := s3api.GenerateSSES3Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	encReader, encIV, err := s3api.CreateSSES3EncryptedReader(bytes.NewReader(plaintext), sseKey)
+	if err != nil {
+		t.Fatalf("encrypt: %v", err)
+	}
+	ciphertext, err := io.ReadAll(encReader)
+	if err != nil {
+		t.Fatalf("read ciphertext: %v", err)
+	}
+
+	sseKey.IV = encIV
+	metadataBytes, err := s3api.SerializeSSES3Metadata(sseKey)
+	if err != nil {
+		t.Fatalf("serialize metadata: %v", err)
+	}
+
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSES3Key: metadataBytes,
+		},
+	}
+
+	// Test full path: MaybeDecryptContent → MaybeDecryptReader → decryptSSES3
+	result, err := MaybeDecryptContent(ciphertext, entry)
+	if err != nil {
+		t.Fatalf("MaybeDecryptContent: %v", err)
+	}
+	if !bytes.Equal(result, plaintext) {
+		t.Errorf("SSE-S3 round-trip failed: got %q, want %q", result, plaintext)
+	}
+}
+
+// --- SSE-KMS integration tests ---
+
+// testKMSProvider is a minimal KMSProvider mock for testing.
+type testKMSProvider struct {
+	keyID     string
+	plaintext []byte // the DEK plaintext returned by Decrypt
+}
+
+func (p *testKMSProvider) GenerateDataKey(_ context.Context, _ *kms.GenerateDataKeyRequest) (*kms.GenerateDataKeyResponse, error) {
+	return nil, nil
+}
+
+func (p *testKMSProvider) Decrypt(_ context.Context, _ *kms.DecryptRequest) (*kms.DecryptResponse, error) {
+	return &kms.DecryptResponse{
+		KeyID:     p.keyID,
+		Plaintext: append([]byte(nil), p.plaintext...), // return a copy
+	}, nil
+}
+
+func (p *testKMSProvider) DescribeKey(_ context.Context, _ *kms.DescribeKeyRequest) (*kms.DescribeKeyResponse, error) {
+	return nil, nil
+}
+
+func (p *testKMSProvider) GetKeyID(_ context.Context, keyIdentifier string) (string, error) {
+	return p.keyID, nil
+}
+
+func (p *testKMSProvider) Close() error { return nil }
+
+func TestMaybeDecryptReader_SSEKMS(t *testing.T) {
+	plaintext := []byte("SSE-KMS encrypted content for testing")
+
+	// Generate a random DEK and IV
+	dek := make([]byte, 32)
+	iv := make([]byte, aes.BlockSize)
+	if _, err := io.ReadFull(rand.Reader, dek); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := io.ReadFull(rand.Reader, iv); err != nil {
+		t.Fatal(err)
+	}
+
+	// Encrypt with AES-CTR (same cipher mode as SSE-KMS)
+	block, err := aes.NewCipher(dek)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ciphertext := make([]byte, len(plaintext))
+	cipher.NewCTR(block, iv).XORKeyStream(ciphertext, plaintext)
+
+	// Set up a mock KMS provider that returns our DEK
+	keyID := "test-kms-key-1"
+	encryptedDEK := []byte("fake-encrypted-dek") // mock doesn't validate
+	kms.SetGlobalKMSProvider(&testKMSProvider{
+		keyID:     keyID,
+		plaintext: dek,
+	})
+	defer kms.SetGlobalKMSProvider(nil)
+
+	// Build serialized KMS metadata
+	kmsMetadata := s3api.SSEKMSMetadata{
+		Algorithm:        s3_constants.SSEAlgorithmKMS,
+		KeyID:            keyID,
+		EncryptedDataKey: base64.StdEncoding.EncodeToString(encryptedDEK),
+		IV:               base64.StdEncoding.EncodeToString(iv),
+	}
+	metadataBytes, err := json.Marshal(kmsMetadata)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSEKMSKey: metadataBytes,
+		},
+	}
+
+	// Test MaybeDecryptReader
+	reader := bytes.NewReader(ciphertext)
+	decrypted, err := MaybeDecryptReader(reader, entry)
+	if err != nil {
+		t.Fatalf("MaybeDecryptReader: %v", err)
+	}
+	result, err := io.ReadAll(decrypted)
+	if err != nil {
+		t.Fatalf("ReadAll: %v", err)
+	}
+	if !bytes.Equal(result, plaintext) {
+		t.Errorf("SSE-KMS round-trip failed: got %q, want %q", result, plaintext)
+	}
+}
+
+func TestMaybeDecryptContent_SSEKMS(t *testing.T) {
+	plaintext := []byte("inline SSE-KMS content")
+
+	dek := make([]byte, 32)
+	iv := make([]byte, aes.BlockSize)
+	if _, err := io.ReadFull(rand.Reader, dek); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := io.ReadFull(rand.Reader, iv); err != nil {
+		t.Fatal(err)
+	}
+
+	block, err := aes.NewCipher(dek)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ciphertext := make([]byte, len(plaintext))
+	cipher.NewCTR(block, iv).XORKeyStream(ciphertext, plaintext)
+
+	keyID := "test-kms-key-2"
+	kms.SetGlobalKMSProvider(&testKMSProvider{
+		keyID:     keyID,
+		plaintext: dek,
+	})
+	defer kms.SetGlobalKMSProvider(nil)
+
+	kmsMetadata := s3api.SSEKMSMetadata{
+		Algorithm:        s3_constants.SSEAlgorithmKMS,
+		KeyID:            keyID,
+		EncryptedDataKey: base64.StdEncoding.EncodeToString([]byte("fake-encrypted-dek")),
+		IV:               base64.StdEncoding.EncodeToString(iv),
+	}
+	metadataBytes, err := json.Marshal(kmsMetadata)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			s3_constants.SeaweedFSSSEKMSKey: metadataBytes,
+		},
+	}
+
+	result, err := MaybeDecryptContent(ciphertext, entry)
+	if err != nil {
+		t.Fatalf("MaybeDecryptContent: %v", err)
+	}
+	if !bytes.Equal(result, plaintext) {
+		t.Errorf("SSE-KMS round-trip failed: got %q, want %q", result, plaintext)
+	}
+}

weed/replication/replicator.go

@@ -8,6 +8,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/repl_util"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -25,6 +26,10 @@ func NewReplicator(sourceConfig util.Configuration, configPrefix string, dataSin
 	source := &source.FilerSource{}
 	source.Initialize(sourceConfig, configPrefix)
 
+	if err := repl_util.InitializeSSEForReplication(source); err != nil {
+		glog.Warningf("SSE initialization failed: %v (encrypted objects may fail to replicate)", err)
+	}
+
 	dataSink.SetSourceFiler(source)
 
 	return &Replicator{

weed/replication/sink/azuresink/azure_sink.go

@@ -138,6 +138,7 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []
 	_, err := appendBlobClient.Create(ctxCreate, nil)
 
 	needsWrite := true
+	freshlyCreated := false
 	if err != nil {
 		if bloberror.HasCode(err, bloberror.BlobAlreadyExists) {
 			// Handle existing blob - check if overwrite is needed and perform it if necessary
@@ -146,9 +147,13 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []
 			if handleErr != nil {
 				return handleErr
 			}
+			// handleExistingBlob recreates the blob when needsWrite is true
+			freshlyCreated = needsWrite
 		} else {
 			return fmt.Errorf("azure create append blob %s/%s: %w", g.container, key, err)
 		}
+	} else {
+		freshlyCreated = true
 	}
 
 	// If we don't need to write (blob is up-to-date), return early
@@ -156,6 +161,23 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []
 		return nil
 	}
 
+	// cleanupOnError deletes a freshly created blob when content write fails,
+	// preventing empty blobs from being left behind.
+	cleanupOnError := func(writeErr error) error {
+		if !freshlyCreated {
+			return writeErr
+		}
+		glog.Warningf("azure sink: cleaning up empty blob %s/%s after write failure: %v", g.container, key, writeErr)
+		ctxCleanup, cancelCleanup := context.WithTimeout(context.Background(), azure.DefaultAzureOpTimeout)
+		defer cancelCleanup()
+		if _, delErr := appendBlobClient.Delete(ctxCleanup, nil); delErr != nil {
+			if !bloberror.HasCode(delErr, bloberror.BlobNotFound) {
+				glog.Warningf("azure sink: failed to clean up blob %s/%s: %v", g.container, key, delErr)
+			}
+		}
+		return writeErr
+	}
+
 	writeFunc := func(data []byte) error {
 		ctxWrite, cancelWrite := context.WithTimeout(context.Background(), azure.DefaultAzureOpTimeout)
 		defer cancelWrite()
@@ -164,11 +186,18 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []
 	}
 
 	if len(entry.Content) > 0 {
-		return writeFunc(entry.Content)
+		content, err := repl_util.MaybeDecryptContent(entry.Content, entry)
+		if err != nil {
+			return cleanupOnError(fmt.Errorf("decrypt inline SSE content: %w", err))
+		}
+		if err := writeFunc(content); err != nil {
+			return cleanupOnError(err)
+		}
+		return nil
 	}
 
-	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
-		return err
+	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc, entry); err != nil {
+		return cleanupOnError(err)
 	}
 
 	return nil

weed/replication/sink/b2sink/b2_sink.go

@@ -2,6 +2,7 @@ package B2Sink
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"github.com/kurin/blazer/b2"
@@ -116,10 +117,14 @@ func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int
 	}
 
 	if len(entry.Content) > 0 {
-		return writeFunc(entry.Content)
+		content, err := repl_util.MaybeDecryptContent(entry.Content, entry)
+		if err != nil {
+			return fmt.Errorf("decrypt inline SSE content: %w", err)
+		}
+		return writeFunc(content)
 	}
 
-	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
+	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc, entry); err != nil {
 		return err
 	}
 

weed/replication/sink/gcssink/gcs_sink.go

@@ -115,23 +115,37 @@ func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []in
 	totalSize := filer.FileSize(entry)
 	chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize))
 
-	wc := g.client.Bucket(g.bucket).Object(key).NewWriter(context.Background())
-	defer wc.Close()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	wc := g.client.Bucket(g.bucket).Object(key).NewWriter(ctx)
 
 	writeFunc := func(data []byte) error {
 		_, writeErr := wc.Write(data)
 		return writeErr
 	}
 
+	var writeErr error
 	if len(entry.Content) > 0 {
-		return writeFunc(entry.Content)
+		content, decErr := repl_util.MaybeDecryptContent(entry.Content, entry)
+		if decErr != nil {
+			writeErr = fmt.Errorf("decrypt inline SSE content: %w", decErr)
+		} else {
+			writeErr = writeFunc(content)
+		}
+	} else {
+		writeErr = repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc, entry)
 	}
 
-	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
-		return err
+	if writeErr != nil {
+		// Cancel the context to abort the GCS upload without touching
+		// any existing object at this key.
+		cancel()
+		wc.Close()
+		return writeErr
 	}
 
-	return nil
+	return wc.Close()
 
 }
 

weed/replication/sink/localsink/local_sink.go

@@ -2,6 +2,7 @@ package localsink
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
@@ -94,36 +95,55 @@ func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signa
 
 	mode := os.FileMode(entry.Attributes.FileMode)
 	shortFileName := util.ToShortFileName(key)
-	if err := os.Remove(shortFileName); err != nil && !os.IsNotExist(err) {
-		return err
-	}
-	dstFile, err := os.OpenFile(shortFileName, os.O_RDWR|os.O_CREATE|os.O_TRUNC, mode)
-	if err != nil {
-		return err
-	}
-	defer dstFile.Close()
 
-	fi, err := dstFile.Stat()
+	// Write to a temp file in the same directory, then atomically rename
+	// on success. This prevents leaving a truncated/empty file if
+	// decryption or chunk copy fails.
+	tmpFile, err := os.CreateTemp(dir, ".seaweedfs-tmp-*")
 	if err != nil {
 		return err
 	}
-	if fi.Mode() != mode {
-		glog.V(4).Infof("Modify file mode: %o -> %o", fi.Mode(), mode)
-		if err := dstFile.Chmod(mode); err != nil {
-			return err
+	tmpName := tmpFile.Name()
+	defer func() {
+		// Clean up temp file on any error (rename removes it on success)
+		if tmpFile != nil {
+			tmpFile.Close()
+			os.Remove(tmpName)
 		}
+	}()
+
+	if err := tmpFile.Chmod(mode); err != nil {
+		return err
 	}
 
 	writeFunc := func(data []byte) error {
-		_, writeErr := dstFile.Write(data)
+		_, writeErr := tmpFile.Write(data)
 		return writeErr
 	}
 
 	if len(entry.Content) > 0 {
-		return writeFunc(entry.Content)
+		content, err := repl_util.MaybeDecryptContent(entry.Content, entry)
+		if err != nil {
+			return fmt.Errorf("decrypt inline SSE content: %w", err)
+		}
+		if err := writeFunc(content); err != nil {
+			return err
+		}
+	} else {
+		if err := repl_util.CopyFromChunkViews(chunkViews, localsink.filerSource, writeFunc, entry); err != nil {
+			return err
+		}
 	}
 
-	if err := repl_util.CopyFromChunkViews(chunkViews, localsink.filerSource, writeFunc); err != nil {
+	// Close before rename so the data is flushed
+	if err := tmpFile.Close(); err != nil {
+		return err
+	}
+	tmpFile = nil // prevent deferred cleanup
+
+	// Atomic rename into final destination
+	if err := os.Rename(tmpName, shortFileName); err != nil {
+		os.Remove(tmpName)
 		return err
 	}
 

weed/replication/sink/s3sink/s3_sink.go

@@ -3,6 +3,7 @@ package S3Sink
 import (
 	"encoding/base64"
 	"fmt"
+	"net/url"
 	"strconv"
 	"strings"
 
@@ -17,6 +18,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/repl_util"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -160,6 +162,14 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
 
 	reader := filer.NewFileReader(s3sink.filerSource, entry)
 
+	// Decrypt SSE-encrypted objects so the destination receives plaintext
+	decryptedReader, err := repl_util.MaybeDecryptReader(reader, entry)
+	if err != nil {
+		repl_util.CloseReader(reader)
+		return fmt.Errorf("decrypt SSE object: %w", err)
+	}
+	defer repl_util.CloseMaybeDecryptedReader(reader, decryptedReader)
+
 	// Create an uploader with the session and custom options
 	uploader := s3manager.NewUploaderWithClient(s3sink.conn, func(u *s3manager.Uploader) {
 		u.PartSize = int64(s3sink.uploaderPartSizeMb * 1024 * 1024)
@@ -188,20 +198,16 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
 		entry.Extended[s3_constants.AmzUserMetaMtime] = []byte(strconv.FormatInt(entry.Attributes.Mtime, 10))
 	}
 	// process tagging
-	tags := ""
-	for k, v := range entry.Extended {
-		if len(tags) > 0 {
-			tags = tags + "&"
-		}
-		tags = tags + k + "=" + string(v)
-	}
+	tags := buildTaggingString(entry.Extended)
 
 	// Upload the file to S3.
 	uploadInput := s3manager.UploadInput{
-		Bucket:  aws.String(s3sink.bucket),
-		Key:     aws.String(key),
-		Body:    reader,
-		Tagging: aws.String(tags),
+		Bucket: aws.String(s3sink.bucket),
+		Key:    aws.String(key),
+		Body:   decryptedReader,
+	}
+	if tags != "" {
+		uploadInput.Tagging = aws.String(tags)
 	}
 	if len(entry.Attributes.Md5) > 0 {
 		uploadInput.ContentMD5 = aws.String(base64.StdEncoding.EncodeToString([]byte(entry.Attributes.Md5)))
@@ -223,3 +229,18 @@ func cleanKey(key string) string {
 	}
 	return key
 }
+
+// buildTaggingString builds the S3 Tagging header value from entry extended metadata.
+// Only keys with the AmzObjectTaggingPrefix ("X-Amz-Tagging-") are included as object
+// tags. The prefix is stripped and values are URL-encoded to produce a valid S3 tagging
+// query string.
+func buildTaggingString(extended map[string][]byte) string {
+	tagValues := url.Values{}
+	for k, v := range extended {
+		if strings.HasPrefix(k, s3_constants.AmzObjectTaggingPrefix) {
+			tagKey := k[len(s3_constants.AmzObjectTaggingPrefix):]
+			tagValues.Set(tagKey, string(v))
+		}
+	}
+	return tagValues.Encode()
+}

weed/replication/sink/s3sink/s3_sink_test.go

@@ -0,0 +1,59 @@
+package S3Sink
+
+import (
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func TestBuildTaggingString_ShouldStripTagPrefix(t *testing.T) {
+	extended := map[string][]byte{
+		s3_constants.AmzObjectTaggingPrefix + "env": []byte("production"),
+	}
+
+	tagging := buildTaggingString(extended)
+
+	if strings.Contains(tagging, s3_constants.AmzObjectTaggingPrefix) {
+		t.Errorf("tagging should not contain storage prefix %q, got %q", s3_constants.AmzObjectTaggingPrefix, tagging)
+	}
+
+	parsed, err := url.ParseQuery(tagging)
+	if err != nil {
+		t.Fatalf("tagging should be valid URL query: %v", err)
+	}
+	if v := parsed.Get("env"); v != "production" {
+		t.Errorf("expected tag env=production, got %q", v)
+	}
+}
+
+func TestBuildTaggingString_ShouldURLEncodeValues(t *testing.T) {
+	extended := map[string][]byte{
+		s3_constants.AmzObjectTaggingPrefix + "path": []byte("/a/b=c&d"),
+	}
+
+	tagging := buildTaggingString(extended)
+
+	parsed, err := url.ParseQuery(tagging)
+	if err != nil {
+		t.Fatalf("tagging should be valid URL query: %v", err)
+	}
+	if v := parsed.Get("path"); v != "/a/b=c&d" {
+		t.Errorf("expected tag value /a/b=c&d after decoding, got %q", v)
+	}
+}
+
+func TestBuildTaggingString_EmptyWhenNoTags(t *testing.T) {
+	extended := map[string][]byte{
+		"Content-Encoding":            []byte("gzip"),
+		s3_constants.AmzUserMetaMtime: []byte("12345"),
+		s3_constants.SeaweedFSSSES3Key: []byte(`{"algorithm":"AES256","encryptedDEK":"abc"}`),
+	}
+
+	tagging := buildTaggingString(extended)
+
+	if tagging != "" {
+		t.Errorf("expected empty tagging when no tag keys present, got %q", tagging)
+	}
+}

weed/server/master_grpc_server_raft.go

@@ -41,7 +41,7 @@ func (ms *MasterServer) RaftListClusterServers(ctx context.Context, req *master_
 		// Add the current server itself (Peers() only returns other peers)
 		resp.ClusterServers = append(resp.ClusterServers, &master_pb.RaftListClusterServersResponse_ClusterServers{
 			Id:       currentServerName,
-			Address:  string(ms.option.Master),
+			Address:  ms.option.Master.ToGrpcAddress(),
 			Suffrage: "Voter",
 			IsLeader: currentServerName == leader,
 		})

weed/server/raft_server.go

@@ -32,6 +32,7 @@ type RaftServerOption struct {
 	DataDir           string
 	Topo              *topology.Topology
 	RaftResumeState   bool
+	SingleMaster      bool
 	HeartbeatInterval time.Duration
 	ElectionTimeout   time.Duration
 	RaftBootstrap     bool
@@ -176,10 +177,38 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) {
 	if err := s.raftServer.LoadSnapshot(); err != nil {
 		return nil, err
 	}
+
+	// In single-master mode resuming state, the log is not empty so the
+	// normal self-join path won't promote to leader. The server will
+	// self-elect after the election timeout, so use a tiny timeout to
+	// make this near-instant, then restore the original after election.
+	fastResume := option.SingleMaster && option.RaftResumeState && !s.raftServer.IsLogEmpty()
+	if fastResume {
+		s.raftServer.SetElectionTimeout(time.Millisecond)
+	}
+
 	if err := s.raftServer.Start(); err != nil {
 		return nil, err
 	}
 
+	if fastResume {
+		go func() {
+			defer s.raftServer.SetElectionTimeout(option.ElectionTimeout)
+			ticker := time.NewTicker(100 * time.Millisecond)
+			defer ticker.Stop()
+			timeout := time.After(option.ElectionTimeout)
+			for s.raftServer.Leader() == "" {
+				select {
+				case <-timeout:
+					glog.Warningf("Fast resume timed out waiting for leader election, restoring election timeout to %v", option.ElectionTimeout)
+					return
+				case <-ticker.C:
+				}
+			}
+			glog.V(0).Infof("Resumed as leader with election timeout restored to %v", option.ElectionTimeout)
+		}()
+	}
+
 	for name, peer := range s.peers {
 		if err := s.raftServer.AddPeer(name, peer.ToGrpcAddress()); err != nil {
 			return nil, err
@@ -273,3 +302,4 @@ func (s *RaftServer) DoJoinCommand() {
 	}
 
 }
+

weed/shell/command_s3_configure.go

@@ -31,8 +31,13 @@ func (c *commandS3Configure) Name() string {
 func (c *commandS3Configure) Help() string {
 	return `configure and apply s3 options for each bucket
 
-	# see the current configuration file content
-	s3.configure
+		# for workflow-oriented user management, prefer:
+		s3.user.create -name=username -generate_credentials
+		s3.user.show -name=username
+		s3.policy.attach -user=username -name=mypolicy
+
+		# see the current configuration file content
+		s3.configure
 
 	# create a new identity with account information
 	s3.configure -user=username -actions=Read,Write,List,Tagging -buckets=bucket-name -policies=policy1,policy2 -access_key=key -secret_key=secret -account_id=id -account_display_name=name -account_email=email@example.com -apply

weed/shell/command_s3_policy.go

@@ -29,10 +29,18 @@ func (c *commandS3Policy) Name() string {
 func (c *commandS3Policy) Help() string {
 	return `manage s3 policies
 
-	# create or update a policy
-	s3.policy -put -name=mypolicy -file=policy.json
+		# preferred exact commands
+		s3.policy.create -name=mypolicy -file=policy.json
+		s3.policy.show -name=mypolicy
+		s3.policy.list
+		s3.policy.delete -name=mypolicy
+		s3.policy.attach -user=alice -name=mypolicy
+		s3.policy.detach -user=alice -name=mypolicy
 
-	# list all policies
+		# create or update a policy
+		s3.policy -put -name=mypolicy -file=policy.json
+
+		# list all policies
 	s3.policy -list
 
 	# get a policy

weed/shell/command_s3_policy_subcommands.go

@@ -0,0 +1,282 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"sort"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/credential"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
+)
+
+func init() {
+	Commands = append(Commands,
+		&s3ShellCommand{
+			name: "s3.policy.create",
+			help: `create or update a managed S3 policy
+
+	s3.policy.create -name photos-rw -file policy.json`,
+			do: runS3PolicyCreateCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.policy.show",
+			help: `show one managed S3 policy
+
+	s3.policy.show -name photos-rw`,
+			do: runS3PolicyShowCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.policy.list",
+			help: `list managed S3 policies`,
+			do:   runS3PolicyListCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.policy.delete",
+			help: `delete a managed S3 policy
+
+	s3.policy.delete -name photos-rw`,
+			do: runS3PolicyDeleteCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.policy.attach",
+			help: `attach a managed S3 policy to a filer-backed user
+
+	s3.policy.attach -user alice -name photos-rw`,
+			do: runS3PolicyAttachCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.policy.detach",
+			help: `detach a managed S3 policy from a filer-backed user
+
+	s3.policy.detach -user alice -name photos-rw`,
+			do: runS3PolicyDetachCommand,
+		},
+	)
+}
+
+func runS3PolicyCreateCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.policy.create", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	name := fs.String("name", "", "policy name")
+	file := fs.String("file", "", "policy file (json)")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+	if *file == "" {
+		return fmt.Errorf("-file is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3PolicyCreate(ctx, store, *name, *file, writer)
+	})
+}
+
+func runS3PolicyShowCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.policy.show", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	name := fs.String("name", "", "policy name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3PolicyShow(ctx, store, *name, writer)
+	})
+}
+
+func runS3PolicyListCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.policy.list", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3PolicyList(ctx, store, writer)
+	})
+}
+
+func runS3PolicyDeleteCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.policy.delete", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	name := fs.String("name", "", "policy name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3PolicyDelete(ctx, store, *name, writer)
+	})
+}
+
+func runS3PolicyAttachCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return runS3PolicyAttachDetachCommand("s3.policy.attach", true, args, commandEnv, writer)
+}
+
+func runS3PolicyDetachCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return runS3PolicyAttachDetachCommand("s3.policy.detach", false, args, commandEnv, writer)
+}
+
+func runS3PolicyAttachDetachCommand(commandName string, attach bool, args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet(commandName, flag.ContinueOnError)
+	fs.SetOutput(writer)
+	user := fs.String("user", "", "user name")
+	name := fs.String("name", "", "policy name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		if attach {
+			return runS3PolicyAttach(ctx, store, *user, *name, writer)
+		}
+		return runS3PolicyDetach(ctx, store, *user, *name, writer)
+	})
+}
+
+func runS3PolicyCreate(ctx context.Context, store s3ShellStore, name, file string, writer io.Writer) error {
+	document, err := loadPolicyDocumentFromFile(file)
+	if err != nil {
+		return err
+	}
+	if err := store.CreatePolicy(ctx, name, *document); err != nil {
+		return err
+	}
+	fmt.Fprintf(writer, "Saved policy %q.\n", name)
+	return nil
+}
+
+func runS3PolicyShow(ctx context.Context, store s3ShellStore, name string, writer io.Writer) error {
+	document, err := store.GetPolicy(ctx, name)
+	if err != nil {
+		return err
+	}
+	if document == nil {
+		return fmt.Errorf("policy %q not found", name)
+	}
+	formatted, err := marshalPolicyDocument(document)
+	if err != nil {
+		return err
+	}
+	fmt.Fprintln(writer, formatted)
+	return nil
+}
+
+func runS3PolicyList(ctx context.Context, store s3ShellStore, writer io.Writer) error {
+	policies, err := store.GetPolicies(ctx)
+	if err != nil {
+		return err
+	}
+	names := make([]string, 0, len(policies))
+	for name := range policies {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	for _, name := range names {
+		fmt.Fprintln(writer, name)
+	}
+	return nil
+}
+
+func runS3PolicyDelete(ctx context.Context, store s3ShellStore, name string, writer io.Writer) error {
+	if err := store.DeletePolicy(ctx, name); err != nil {
+		if s3ShellIsNotFound(err) {
+			return fmt.Errorf("policy %q not found", name)
+		}
+		return err
+	}
+	fmt.Fprintf(writer, "Deleted policy %q.\n", name)
+	return nil
+}
+
+func runS3PolicyAttach(ctx context.Context, store s3ShellStore, username, policyName string, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+	if err := ensureMutableIdentity(identity, "attach policies to"); err != nil {
+		return err
+	}
+	if err := store.AttachUserPolicy(ctx, username, policyName); err != nil {
+		if errors.Is(err, credential.ErrPolicyNotFound) {
+			return fmt.Errorf("policy %q not found", policyName)
+		}
+		return err
+	}
+	fmt.Fprintf(writer, "Attached policy %q to user %q.\n", policyName, username)
+	return nil
+}
+
+func runS3PolicyDetach(ctx context.Context, store s3ShellStore, username, policyName string, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+	if err := ensureMutableIdentity(identity, "detach policies from"); err != nil {
+		return err
+	}
+	if err := store.DetachUserPolicy(ctx, username, policyName); err != nil {
+		if errors.Is(err, credential.ErrPolicyNotAttached) {
+			return fmt.Errorf("policy %q is not attached to user %q", policyName, username)
+		}
+		return err
+	}
+	fmt.Fprintf(writer, "Detached policy %q from user %q.\n", policyName, username)
+	return nil
+}
+
+func loadPolicyDocumentFromFile(file string) (*policy_engine.PolicyDocument, error) {
+	data, err := os.ReadFile(file)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read policy file: %v", err)
+	}
+	var document policy_engine.PolicyDocument
+	if err := json.Unmarshal(data, &document); err != nil {
+		return nil, fmt.Errorf("invalid policy json: %v", err)
+	}
+	return &document, nil
+}
+
+func marshalPolicyDocument(document *policy_engine.PolicyDocument) (string, error) {
+	data, err := json.MarshalIndent(document, "", "  ")
+	if err != nil {
+		return "", err
+	}
+	return strings.TrimSpace(string(data)), nil
+}

weed/shell/command_s3_user_subcommands.go

@@ -0,0 +1,712 @@
+package shell
+
+import (
+	"context"
+	"crypto/rand"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"sort"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/credential"
+	iamgrpc "github.com/seaweedfs/seaweedfs/weed/credential/grpc"
+	weediam "github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+var errS3AccessKeyInUse = errors.New("access key already in use")
+
+type s3ShellCommand struct {
+	name string
+	help string
+	do   func([]string, *CommandEnv, io.Writer) error
+}
+
+func (c *s3ShellCommand) Name() string { return c.name }
+
+func (c *s3ShellCommand) Help() string { return c.help }
+
+func (c *s3ShellCommand) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return c.do(args, commandEnv, writer)
+}
+
+func (c *s3ShellCommand) HasTag(CommandTag) bool { return false }
+
+type s3ShellStore interface {
+	CreateUser(ctx context.Context, identity *iam_pb.Identity) error
+	GetUser(ctx context.Context, username string) (*iam_pb.Identity, error)
+	UpdateUser(ctx context.Context, username string, identity *iam_pb.Identity) error
+	DeleteUser(ctx context.Context, username string) error
+	ListUsers(ctx context.Context) ([]string, error)
+	GetUserByAccessKey(ctx context.Context, accessKey string) (*iam_pb.Identity, error)
+	CreateAccessKey(ctx context.Context, username string, credential *iam_pb.Credential) error
+	DeleteAccessKey(ctx context.Context, username string, accessKey string) error
+	GetPolicy(ctx context.Context, name string) (*policy_engine.PolicyDocument, error)
+	GetPolicies(ctx context.Context) (map[string]policy_engine.PolicyDocument, error)
+	CreatePolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error
+	DeletePolicy(ctx context.Context, name string) error
+	AttachUserPolicy(ctx context.Context, username string, policyName string) error
+	DetachUserPolicy(ctx context.Context, username string, policyName string) error
+	ListAttachedUserPolicies(ctx context.Context, username string) ([]string, error)
+}
+
+type s3UserCreateOptions struct {
+	name                string
+	accessKey           string
+	secretKey           string
+	generateCredentials bool
+	accountID           string
+	displayName         string
+	email               string
+}
+
+type s3AccessKeyCreateOptions struct {
+	username            string
+	accessKey           string
+	secretKey           string
+	generateCredentials bool
+}
+
+func init() {
+	Commands = append(Commands,
+		&s3ShellCommand{
+			name: "s3.user.list",
+			help: `list S3 users with source and status`,
+			do:   runS3UserListCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.show",
+			help: `show one S3 user's details
+
+	s3.user.show -name alice`,
+			do: runS3UserShowCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.create",
+			help: `create a filer-backed S3 user
+
+	s3.user.create -name alice -generate_credentials
+	s3.user.create -name alice -access_key AKIA... -secret_key secret
+	s3.user.create -name alice -email alice@example.com -display_name "Alice"`,
+			do: runS3UserCreateCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.delete",
+			help: `delete a filer-backed S3 user
+
+	s3.user.delete -name alice`,
+			do: runS3UserDeleteCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.enable",
+			help: `enable a filer-backed S3 user
+
+	s3.user.enable -name alice`,
+			do: runS3UserEnableCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.disable",
+			help: `disable a filer-backed S3 user
+
+	s3.user.disable -name alice`,
+			do: runS3UserDisableCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.accesskey.list",
+			help: `list access keys for a filer-backed S3 user
+
+	s3.user.accesskey.list -user alice`,
+			do: runS3UserAccessKeyListCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.accesskey.create",
+			help: `create an access key for a filer-backed S3 user
+
+	s3.user.accesskey.create -user alice -generate_credentials
+	s3.user.accesskey.create -user alice -access_key AKIA... -secret_key secret`,
+			do: runS3UserAccessKeyCreateCommand,
+		},
+		&s3ShellCommand{
+			name: "s3.user.accesskey.delete",
+			help: `delete an access key for a filer-backed S3 user
+
+	s3.user.accesskey.delete -user alice -access_key AKIA...`,
+			do: runS3UserAccessKeyDeleteCommand,
+		},
+	)
+}
+
+func runS3UserListCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.list", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserList(ctx, store, writer)
+	})
+}
+
+func runS3UserShowCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.show", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	name := fs.String("name", "", "user name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserShow(ctx, store, *name, writer)
+	})
+}
+
+func runS3UserCreateCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.create", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	opts := s3UserCreateOptions{}
+	fs.StringVar(&opts.name, "name", "", "user name")
+	fs.StringVar(&opts.accessKey, "access_key", "", "access key to create")
+	fs.StringVar(&opts.secretKey, "secret_key", "", "secret key to create")
+	fs.BoolVar(&opts.generateCredentials, "generate_credentials", false, "generate an initial access key pair")
+	fs.StringVar(&opts.accountID, "account_id", "", "account id")
+	fs.StringVar(&opts.displayName, "display_name", "", "account display name")
+	fs.StringVar(&opts.email, "email", "", "account email")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if opts.name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserCreate(ctx, store, opts, writer)
+	})
+}
+
+func runS3UserDeleteCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.delete", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	name := fs.String("name", "", "user name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserDelete(ctx, store, *name, writer)
+	})
+}
+
+func runS3UserEnableCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return runS3UserStatusCommand("s3.user.enable", false, args, commandEnv, writer)
+}
+
+func runS3UserDisableCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return runS3UserStatusCommand("s3.user.disable", true, args, commandEnv, writer)
+}
+
+func runS3UserStatusCommand(commandName string, disabled bool, args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet(commandName, flag.ContinueOnError)
+	fs.SetOutput(writer)
+	name := fs.String("name", "", "user name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserSetDisabled(ctx, store, *name, disabled, writer)
+	})
+}
+
+func runS3UserAccessKeyListCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.accesskey.list", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	user := fs.String("user", "", "user name")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserAccessKeyList(ctx, store, *user, writer)
+	})
+}
+
+func runS3UserAccessKeyCreateCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.accesskey.create", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	opts := s3AccessKeyCreateOptions{}
+	fs.StringVar(&opts.username, "user", "", "user name")
+	fs.StringVar(&opts.accessKey, "access_key", "", "access key to create")
+	fs.StringVar(&opts.secretKey, "secret_key", "", "secret key to create")
+	fs.BoolVar(&opts.generateCredentials, "generate_credentials", false, "generate an access key pair")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if opts.username == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserAccessKeyCreate(ctx, store, opts, writer)
+	})
+}
+
+func runS3UserAccessKeyDeleteCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet("s3.user.accesskey.delete", flag.ContinueOnError)
+	fs.SetOutput(writer)
+	user := fs.String("user", "", "user name")
+	accessKey := fs.String("access_key", "", "access key to delete")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+	if *accessKey == "" {
+		return fmt.Errorf("-access_key is required")
+	}
+
+	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
+		return runS3UserAccessKeyDelete(ctx, store, *user, *accessKey, writer)
+	})
+}
+
+func runS3UserList(ctx context.Context, store s3ShellStore, writer io.Writer) error {
+	usernames, err := store.ListUsers(ctx)
+	if err != nil {
+		return err
+	}
+	sort.Strings(usernames)
+
+	tw := tabwriter.NewWriter(writer, 0, 0, 2, ' ', 0)
+	fmt.Fprintln(tw, "NAME\tSOURCE\tSTATUS\tACCESS KEYS\tPOLICIES")
+	for _, username := range usernames {
+		identity, err := store.GetUser(ctx, username)
+		if err != nil {
+			return err
+		}
+		fmt.Fprintf(tw, "%s\t%s\t%s\t%d\t%d\n",
+			identity.Name,
+			s3IdentitySource(identity),
+			s3IdentityStatus(identity),
+			len(identity.Credentials),
+			len(identity.PolicyNames),
+		)
+	}
+	return tw.Flush()
+}
+
+func runS3UserShow(ctx context.Context, store s3ShellStore, username string, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+	writeS3IdentityDetails(writer, identity)
+	return nil
+}
+
+func runS3UserCreate(ctx context.Context, store s3ShellStore, opts s3UserCreateOptions, writer io.Writer) error {
+	existing, err := store.GetUser(ctx, opts.name)
+	switch {
+	case err == nil && existing != nil:
+		if existing.IsStatic {
+			return fmt.Errorf("user %q already exists in -s3.config; edit the static config for bootstrap users", opts.name)
+		}
+		return fmt.Errorf("user %q already exists", opts.name)
+	case err != nil && !s3ShellIsNotFound(err):
+		return err
+	}
+
+	identity := &iam_pb.Identity{Name: opts.name}
+	if opts.accountID != "" || opts.displayName != "" || opts.email != "" {
+		identity.Account = &iam_pb.Account{
+			Id:           opts.accountID,
+			DisplayName:  opts.displayName,
+			EmailAddress: opts.email,
+		}
+	}
+
+	createdCredential, err := s3BuildCredentialForCreate(ctx, store, opts.accessKey, opts.secretKey, opts.generateCredentials || opts.accessKey != "" || opts.secretKey != "")
+	if err != nil {
+		return err
+	}
+	if createdCredential != nil {
+		identity.Credentials = []*iam_pb.Credential{createdCredential}
+	}
+
+	if err := store.CreateUser(ctx, identity); err != nil {
+		if s3ShellIsAlreadyExists(err) {
+			return fmt.Errorf("user %q already exists", opts.name)
+		}
+		return err
+	}
+
+	fmt.Fprintf(writer, "Created user %q.\n", opts.name)
+	if createdCredential != nil {
+		writeCreatedCredential(writer, createdCredential)
+	}
+	return nil
+}
+
+func runS3UserDelete(ctx context.Context, store s3ShellStore, username string, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+	if err := ensureMutableIdentity(identity, "delete"); err != nil {
+		return err
+	}
+	if err := store.DeleteUser(ctx, username); err != nil {
+		if s3ShellIsNotFound(err) {
+			return fmt.Errorf("user %q not found", username)
+		}
+		return err
+	}
+	fmt.Fprintf(writer, "Deleted user %q.\n", username)
+	return nil
+}
+
+func runS3UserSetDisabled(ctx context.Context, store s3ShellStore, username string, disabled bool, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+	if err := ensureMutableIdentity(identity, ternary(disabled, "disable", "enable")); err != nil {
+		return err
+	}
+	if identity.Disabled == disabled {
+		fmt.Fprintf(writer, "User %q is already %s.\n", username, s3IdentityStatus(identity))
+		return nil
+	}
+
+	updated := proto.Clone(identity).(*iam_pb.Identity)
+	updated.Disabled = disabled
+	if err := store.UpdateUser(ctx, username, updated); err != nil {
+		return err
+	}
+	if disabled {
+		fmt.Fprintf(writer, "Disabled user %q.\n", username)
+	} else {
+		fmt.Fprintf(writer, "Enabled user %q.\n", username)
+	}
+	return nil
+}
+
+func runS3UserAccessKeyList(ctx context.Context, store s3ShellStore, username string, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+
+	tw := tabwriter.NewWriter(writer, 0, 0, 2, ' ', 0)
+	fmt.Fprintln(tw, "USER\tACCESS KEY\tSTATUS")
+	for _, credential := range sortedCredentials(identity.Credentials) {
+		fmt.Fprintf(tw, "%s\t%s\t%s\n", username, credential.AccessKey, s3CredentialStatus(credential))
+	}
+	if len(identity.Credentials) == 0 {
+		fmt.Fprintf(tw, "%s\t%s\t%s\n", username, "-", "-")
+	}
+	return tw.Flush()
+}
+
+func runS3UserAccessKeyCreate(ctx context.Context, store s3ShellStore, opts s3AccessKeyCreateOptions, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, opts.username)
+	if err != nil {
+		return err
+	}
+	if err := ensureMutableIdentity(identity, "create access keys for"); err != nil {
+		return err
+	}
+
+	createdCredential, err := s3BuildCredentialForCreate(ctx, store, opts.accessKey, opts.secretKey, opts.generateCredentials || opts.accessKey != "" || opts.secretKey != "")
+	if err != nil {
+		return err
+	}
+	if createdCredential == nil {
+		return fmt.Errorf("set -generate_credentials or provide -access_key and/or -secret_key")
+	}
+
+	if err := store.CreateAccessKey(ctx, opts.username, createdCredential); err != nil {
+		if s3ShellIsAlreadyExists(err) {
+			return fmt.Errorf("access key %q is already in use", createdCredential.AccessKey)
+		}
+		return err
+	}
+
+	fmt.Fprintf(writer, "Created access key for user %q.\n", opts.username)
+	writeCreatedCredential(writer, createdCredential)
+	return nil
+}
+
+func runS3UserAccessKeyDelete(ctx context.Context, store s3ShellStore, username string, accessKey string, writer io.Writer) error {
+	identity, err := getS3User(ctx, store, username)
+	if err != nil {
+		return err
+	}
+	if err := ensureMutableIdentity(identity, "delete access keys for"); err != nil {
+		return err
+	}
+	if err := store.DeleteAccessKey(ctx, username, accessKey); err != nil {
+		if s3ShellIsNotFound(err) {
+			return fmt.Errorf("access key %q not found for user %q", accessKey, username)
+		}
+		return err
+	}
+	fmt.Fprintf(writer, "Deleted access key %q for user %q.\n", accessKey, username)
+	return nil
+}
+
+func withS3ShellStore(commandEnv *CommandEnv, fn func(ctx context.Context, store s3ShellStore) error) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	store := &iamgrpc.IamGrpcStore{}
+	store.SetFilerAddressFunc(func() pb.ServerAddress {
+		return commandEnv.option.FilerAddress
+	}, commandEnv.option.GrpcDialOption)
+
+	return fn(ctx, store)
+}
+
+func ensureMutableIdentity(identity *iam_pb.Identity, action string) error {
+	if identity != nil && identity.IsStatic {
+		return fmt.Errorf("cannot %s user %q because it comes from -s3.config; edit the static config for bootstrap users", action, identity.Name)
+	}
+	return nil
+}
+
+func getS3User(ctx context.Context, store s3ShellStore, username string) (*iam_pb.Identity, error) {
+	identity, err := store.GetUser(ctx, username)
+	if err != nil {
+		if s3ShellIsNotFound(err) {
+			return nil, fmt.Errorf("user %q not found", username)
+		}
+		return nil, err
+	}
+	return identity, nil
+}
+
+func s3BuildCredentialForCreate(ctx context.Context, store s3ShellStore, accessKey, secretKey string, shouldCreate bool) (*iam_pb.Credential, error) {
+	if !shouldCreate {
+		return nil, nil
+	}
+
+	if accessKey == "" {
+		var err error
+		accessKey, err = generateUniqueAccessKey(ctx, store)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		if err := ensureAccessKeyAvailable(ctx, store, accessKey); err != nil {
+			return nil, err
+		}
+	}
+	if secretKey == "" {
+		var err error
+		secretKey, err = weediam.GenerateSecretAccessKey()
+		if err != nil {
+			return nil, fmt.Errorf("generate secret key: %w", err)
+		}
+	}
+
+	return &iam_pb.Credential{
+		AccessKey: accessKey,
+		SecretKey: secretKey,
+		Status:    weediam.AccessKeyStatusActive,
+	}, nil
+}
+
+func ensureAccessKeyAvailable(ctx context.Context, store s3ShellStore, accessKey string) error {
+	_, err := store.GetUserByAccessKey(ctx, accessKey)
+	if err == nil {
+		return fmt.Errorf("%w: %s", errS3AccessKeyInUse, accessKey)
+	}
+	if s3ShellIsNotFound(err) {
+		return nil
+	}
+	return err
+}
+
+func generateUniqueAccessKey(ctx context.Context, store s3ShellStore) (string, error) {
+	for range 16 {
+		accessKey, err := generateAccessKey()
+		if err != nil {
+			return "", err
+		}
+		if err := ensureAccessKeyAvailable(ctx, store, accessKey); err == nil {
+			return accessKey, nil
+		} else if !errors.Is(err, errS3AccessKeyInUse) {
+			return "", err
+		}
+	}
+	return "", fmt.Errorf("failed to generate a unique access key")
+}
+
+func generateAccessKey() (string, error) {
+	const charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+	var raw [20]byte
+	if _, err := rand.Read(raw[:]); err != nil {
+		return "", fmt.Errorf("generate access key: %w", err)
+	}
+	for i := range raw {
+		raw[i] = charset[int(raw[i])%len(charset)]
+	}
+	return string(raw[:]), nil
+}
+
+func s3IdentitySource(identity *iam_pb.Identity) string {
+	if identity != nil && identity.IsStatic {
+		return "static"
+	}
+	return "dynamic"
+}
+
+func s3IdentityStatus(identity *iam_pb.Identity) string {
+	if identity != nil && identity.Disabled {
+		return "disabled"
+	}
+	return "enabled"
+}
+
+func s3CredentialStatus(credential *iam_pb.Credential) string {
+	if credential == nil || credential.Status == "" {
+		return weediam.AccessKeyStatusActive
+	}
+	return credential.Status
+}
+
+func sortedCredentials(credentials []*iam_pb.Credential) []*iam_pb.Credential {
+	out := append([]*iam_pb.Credential(nil), credentials...)
+	sort.Slice(out, func(i, j int) bool {
+		return out[i].AccessKey < out[j].AccessKey
+	})
+	return out
+}
+
+func writeCreatedCredential(writer io.Writer, credential *iam_pb.Credential) {
+	fmt.Fprintf(writer, "Access Key: %s\n", credential.AccessKey)
+	fmt.Fprintf(writer, "Secret Key: %s\n", credential.SecretKey)
+}
+
+func writeS3IdentityDetails(writer io.Writer, identity *iam_pb.Identity) {
+	fmt.Fprintf(writer, "Name: %s\n", identity.Name)
+	fmt.Fprintf(writer, "Source: %s\n", s3IdentitySource(identity))
+	fmt.Fprintf(writer, "Status: %s\n", s3IdentityStatus(identity))
+	if identity.Account != nil {
+		if identity.Account.Id != "" {
+			fmt.Fprintf(writer, "Account ID: %s\n", identity.Account.Id)
+		}
+		if identity.Account.DisplayName != "" {
+			fmt.Fprintf(writer, "Display Name: %s\n", identity.Account.DisplayName)
+		}
+		if identity.Account.EmailAddress != "" {
+			fmt.Fprintf(writer, "Email: %s\n", identity.Account.EmailAddress)
+		}
+	}
+
+	if len(identity.Credentials) == 0 {
+		fmt.Fprintln(writer, "Access Keys: none")
+	} else {
+		fmt.Fprintln(writer, "Access Keys:")
+		for _, credential := range sortedCredentials(identity.Credentials) {
+			fmt.Fprintf(writer, "  %s (%s)\n", credential.AccessKey, s3CredentialStatus(credential))
+		}
+	}
+
+	if len(identity.PolicyNames) == 0 {
+		fmt.Fprintln(writer, "Policies: none")
+	} else {
+		policies := append([]string(nil), identity.PolicyNames...)
+		sort.Strings(policies)
+		fmt.Fprintf(writer, "Policies: %s\n", strings.Join(policies, ", "))
+	}
+
+	if len(identity.Actions) == 0 {
+		fmt.Fprintln(writer, "Actions: none")
+	} else {
+		actions := append([]string(nil), identity.Actions...)
+		sort.Strings(actions)
+		fmt.Fprintf(writer, "Actions: %s\n", strings.Join(actions, ", "))
+	}
+}
+
+func s3ShellIsNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, credential.ErrUserNotFound) ||
+		errors.Is(err, credential.ErrAccessKeyNotFound) ||
+		errors.Is(err, credential.ErrPolicyNotFound) {
+		return true
+	}
+	if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
+		return true
+	}
+	return false
+}
+
+func s3ShellIsAlreadyExists(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, credential.ErrUserAlreadyExists) {
+		return true
+	}
+	if st, ok := status.FromError(err); ok && st.Code() == codes.AlreadyExists {
+		return true
+	}
+	return false
+}
+
+func ternary[T any](cond bool, onTrue, onFalse T) T {
+	if cond {
+		return onTrue
+	}
+	return onFalse
+}

weed/shell/command_s3_user_subcommands_test.go

@@ -0,0 +1,191 @@
+package shell
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/credential/memory"
+	weediam "github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
+)
+
+func newTestS3ShellStore(t *testing.T) *memory.MemoryStore {
+	t.Helper()
+
+	store := &memory.MemoryStore{}
+	if err := store.Initialize(nil, ""); err != nil {
+		t.Fatalf("initialize memory store: %v", err)
+	}
+	return store
+}
+
+func TestRunS3UserCreateGeneratesCredentials(t *testing.T) {
+	store := newTestS3ShellStore(t)
+
+	var out bytes.Buffer
+	err := runS3UserCreate(context.Background(), store, s3UserCreateOptions{
+		name:                "alice",
+		generateCredentials: true,
+		email:               "alice@example.com",
+	}, &out)
+	if err != nil {
+		t.Fatalf("runS3UserCreate: %v", err)
+	}
+
+	user, err := store.GetUser(context.Background(), "alice")
+	if err != nil {
+		t.Fatalf("GetUser: %v", err)
+	}
+	if len(user.Credentials) != 1 {
+		t.Fatalf("expected 1 credential, got %d", len(user.Credentials))
+	}
+	if got := user.Credentials[0].Status; got != weediam.AccessKeyStatusActive {
+		t.Fatalf("expected %q status, got %q", weediam.AccessKeyStatusActive, got)
+	}
+	if user.Account == nil || user.Account.EmailAddress != "alice@example.com" {
+		t.Fatalf("expected account email to be set, got %+v", user.Account)
+	}
+
+	output := out.String()
+	if !strings.Contains(output, `Created user "alice".`) {
+		t.Fatalf("expected created message, got %q", output)
+	}
+	if !strings.Contains(output, "Access Key:") || !strings.Contains(output, "Secret Key:") {
+		t.Fatalf("expected credential output, got %q", output)
+	}
+}
+
+func TestRunS3UserSetDisabledRejectsStaticUser(t *testing.T) {
+	store := newTestS3ShellStore(t)
+	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
+		Name:     "bootstrap-admin",
+		IsStatic: true,
+	}); err != nil {
+		t.Fatalf("CreateUser: %v", err)
+	}
+
+	err := runS3UserSetDisabled(context.Background(), store, "bootstrap-admin", true, &bytes.Buffer{})
+	if err == nil {
+		t.Fatal("expected static user mutation to fail")
+	}
+	if !strings.Contains(err.Error(), "-s3.config") {
+		t.Fatalf("expected static user guidance, got %v", err)
+	}
+}
+
+func TestRunS3UserListIncludesSourceAndStatus(t *testing.T) {
+	store := newTestS3ShellStore(t)
+	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
+		Name: "alice",
+		Credentials: []*iam_pb.Credential{
+			{AccessKey: "ALICEKEY", SecretKey: "secret", Status: weediam.AccessKeyStatusActive},
+		},
+	}); err != nil {
+		t.Fatalf("CreateUser alice: %v", err)
+	}
+	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
+		Name:     "bootstrap-admin",
+		IsStatic: true,
+		Disabled: true,
+	}); err != nil {
+		t.Fatalf("CreateUser bootstrap-admin: %v", err)
+	}
+
+	var out bytes.Buffer
+	if err := runS3UserList(context.Background(), store, &out); err != nil {
+		t.Fatalf("runS3UserList: %v", err)
+	}
+
+	output := out.String()
+	if !strings.Contains(output, "NAME") || !strings.Contains(output, "SOURCE") || !strings.Contains(output, "STATUS") {
+		t.Fatalf("expected header, got %q", output)
+	}
+	if !strings.Contains(output, "alice") || !strings.Contains(output, "dynamic") || !strings.Contains(output, "enabled") {
+		t.Fatalf("expected dynamic user row, got %q", output)
+	}
+	if !strings.Contains(output, "bootstrap-admin") || !strings.Contains(output, "static") || !strings.Contains(output, "disabled") {
+		t.Fatalf("expected static user row, got %q", output)
+	}
+}
+
+func TestRunS3UserAccessKeyCreateGeneratesCredential(t *testing.T) {
+	store := newTestS3ShellStore(t)
+	if err := store.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil {
+		t.Fatalf("CreateUser: %v", err)
+	}
+
+	var out bytes.Buffer
+	err := runS3UserAccessKeyCreate(context.Background(), store, s3AccessKeyCreateOptions{
+		username:            "alice",
+		generateCredentials: true,
+	}, &out)
+	if err != nil {
+		t.Fatalf("runS3UserAccessKeyCreate: %v", err)
+	}
+
+	user, err := store.GetUser(context.Background(), "alice")
+	if err != nil {
+		t.Fatalf("GetUser: %v", err)
+	}
+	if len(user.Credentials) != 1 {
+		t.Fatalf("expected 1 credential, got %d", len(user.Credentials))
+	}
+	if !strings.Contains(out.String(), `Created access key for user "alice".`) {
+		t.Fatalf("expected create message, got %q", out.String())
+	}
+}
+
+func TestRunS3PolicyAttachUpdatesUserPolicies(t *testing.T) {
+	store := newTestS3ShellStore(t)
+	if err := store.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil {
+		t.Fatalf("CreateUser: %v", err)
+	}
+	if err := store.CreatePolicy(context.Background(), "photos-rw", policy_engine.PolicyDocument{
+		Version: "2012-10-17",
+	}); err != nil {
+		t.Fatalf("CreatePolicy: %v", err)
+	}
+
+	var out bytes.Buffer
+	if err := runS3PolicyAttach(context.Background(), store, "alice", "photos-rw", &out); err != nil {
+		t.Fatalf("runS3PolicyAttach: %v", err)
+	}
+
+	user, err := store.GetUser(context.Background(), "alice")
+	if err != nil {
+		t.Fatalf("GetUser: %v", err)
+	}
+	if len(user.PolicyNames) != 1 || user.PolicyNames[0] != "photos-rw" {
+		t.Fatalf("expected attached policy, got %v", user.PolicyNames)
+	}
+	if !strings.Contains(out.String(), `Attached policy "photos-rw" to user "alice".`) {
+		t.Fatalf("expected attach message, got %q", out.String())
+	}
+}
+
+func TestRunS3PolicyDetachRejectsStaticUser(t *testing.T) {
+	store := newTestS3ShellStore(t)
+	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
+		Name:        "bootstrap-admin",
+		IsStatic:    true,
+		PolicyNames: []string{"photos-rw"},
+	}); err != nil {
+		t.Fatalf("CreateUser: %v", err)
+	}
+	if err := store.CreatePolicy(context.Background(), "photos-rw", policy_engine.PolicyDocument{
+		Version: "2012-10-17",
+	}); err != nil {
+		t.Fatalf("CreatePolicy: %v", err)
+	}
+
+	err := runS3PolicyDetach(context.Background(), store, "bootstrap-admin", "photos-rw", &bytes.Buffer{})
+	if err == nil {
+		t.Fatal("expected static user mutation to fail")
+	}
+	if !strings.Contains(err.Error(), "-s3.config") {
+		t.Fatalf("expected static user guidance, got %v", err)
+	}
+}

weed/shell/command_volume_fix_replication.go

@@ -1,7 +1,6 @@
 package shell
 
 import (
-	"context"
 	"flag"
 	"fmt"
 	"io"
@@ -16,11 +15,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle_map"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
-	"github.com/seaweedfs/seaweedfs/weed/util"
 
-	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 )
 
@@ -390,30 +386,10 @@ func (c *commandVolumeFixReplication) fixOneUnderReplicatedVolume(commandEnv *Co
 				return true, nil
 			}
 
-			err := operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dst.dataNode), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-				stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
-					VolumeId:       replica.info.Id,
-					SourceDataNode: string(pb.NewServerAddressFromDataNode(replica.location.dataNode)),
-				})
-				if replicateErr != nil {
-					return fmt.Errorf("copying from %s => %s : %v", replica.location.dataNode.Id, dst.dataNode.Id, replicateErr)
-				}
-				for {
-					resp, recvErr := stream.Recv()
-					if recvErr != nil {
-						if recvErr == io.EOF {
-							break
-						} else {
-							return recvErr
-						}
-					}
-					if resp.ProcessedBytes > 0 {
-						fmt.Fprintf(writer, "volume %d processed %s bytes\n", replica.info.Id, util.BytesToHumanReadable(uint64(resp.ProcessedBytes)))
-					}
-				}
-
-				return nil
-			})
+			err := replicateVolumeToServer(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(replica.info.Id),
+				pb.NewServerAddressFromDataNode(replica.location.dataNode),
+				pb.NewServerAddressFromDataNode(dst.dataNode),
+				replica.info.DiskType)
 
 			if err != nil {
 				return false, err

weed/shell/command_volume_move.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -240,3 +241,47 @@ func markVolumeReplicasWritable(grpcDialOption grpc.DialOption, volumeId needle.
 	}
 	return nil
 }
+
+// replicateVolumeToServer copies a volume from sourceAddress to targetAddress via the VolumeCopy gRPC stream.
+func replicateVolumeToServer(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceAddress, targetAddress pb.ServerAddress, diskType string) error {
+	return operation.WithVolumeServerClient(false, targetAddress, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+		stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
+			VolumeId:       uint32(volumeId),
+			SourceDataNode: string(sourceAddress),
+			DiskType:       diskType,
+		})
+		if replicateErr != nil {
+			return replicateErr
+		}
+		for {
+			resp, recvErr := stream.Recv()
+			if recvErr != nil {
+				if recvErr == io.EOF {
+					break
+				}
+				return recvErr
+			}
+			if resp.ProcessedBytes > 0 {
+				fmt.Fprintf(writer, "volume %d processed %s bytes\n", volumeId, util.BytesToHumanReadable(uint64(resp.ProcessedBytes)))
+			}
+		}
+		return nil
+	})
+}
+
+// configureVolumeReplication sets the replication setting on a volume at the given server.
+func configureVolumeReplication(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, targetAddress pb.ServerAddress, replicationString string) error {
+	return operation.WithVolumeServerClient(false, targetAddress, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+		resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
+			VolumeId:    uint32(volumeId),
+			Replication: replicationString,
+		})
+		if configureErr != nil {
+			return configureErr
+		}
+		if resp.Error != "" {
+			return errors.New(resp.Error)
+		}
+		return nil
+	})
+}

weed/shell/command_volume_tier_move.go

@@ -1,8 +1,6 @@
 package shell
 
 import (
-	"context"
-	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -13,11 +11,10 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 
-	"github.com/seaweedfs/seaweedfs/weed/operation"
-	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 )
 
@@ -47,8 +44,12 @@ func (c *commandVolumeTierMove) Help() string {
 
 	volume.tier.move -fromDiskType=hdd -toDiskType=ssd [-collectionPattern=""] [-fullPercent=95] [-quietFor=1h] [-parallelLimit=4] [-toReplication=XYZ]
 
-	Even if the volume is replicated, only one replica will be changed and the rest replicas will be dropped.
-	So "volume.fix.replication" and "volume.balance" should be followed.
+	The command ensures the target replication is fully achieved on the destination tier
+	before deleting old replicas. This prevents data loss if a destination disk fails
+	before replication repair completes.
+
+	When -toReplication is specified, the moved volume is reconfigured with the new
+	replication setting. Otherwise, the volume's existing replication is preserved.
 
 	Note:
 		Use -collectionPattern="_default" to match only the default collection (volumes with no collection name).
@@ -271,36 +272,201 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i
 
 	// If move is successful and replication is not empty, alter moved volume's replication setting
 	if *replicationString != "" {
-		err = operation.WithVolumeServerClient(false, newAddress, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-			resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
-				VolumeId:    uint32(vid),
-				Replication: *replicationString,
-			})
-			if configureErr != nil {
-				return configureErr
-			}
-			if resp.Error != "" {
-				return errors.New(resp.Error)
-			}
-			return nil
-		})
-		if err != nil {
-			glog.Errorf("update volume %d replication on %s: %v", vid, locations[0].Url, err)
+		if err = configureVolumeReplication(commandEnv.option.GrpcDialOption, vid, newAddress, *replicationString); err != nil {
+			// LiveMoveVolume already deleted sourceVolumeServer; mark surviving
+			// old replicas writable before aborting so the volume stays accessible.
+			restoreSurvivingReplicasWritable(commandEnv, vid, locations, sourceVolumeServer)
+			return fmt.Errorf("configure replication %s on volume %d at %s: %v", *replicationString, vid, newAddress, err)
 		}
 	}
 
-	// remove the remaining replicas
+	// Ensure the required number of replicas exist on the target tier BEFORE
+	// deleting old replicas to avoid data-loss risk.
+	// Use the explicit -toReplication if given, otherwise preserve the volume's
+	// existing replication from the source tier.
+	preserveServers, replicateErr := c.ensureReplicationFulfilled(commandEnv, writer, vid, toDiskType, dst, *replicationString)
+	if replicateErr != nil {
+		// Replication not fully achieved — do NOT delete old replicas.
+		restoreSurvivingReplicasWritable(commandEnv, vid, locations, sourceVolumeServer)
+		return fmt.Errorf("volume %d moved to %s but failed to fulfill replication, old replicas preserved: %v", vid, dst.dataNode.Id, replicateErr)
+	}
+
+	// Mark preserved pre-existing target-tier replicas as writable.
+	// They were marked read-only at the start of the move and would otherwise
+	// stay read-only since we're keeping rather than deleting them.
 	for _, loc := range locations {
-		if loc.Url != dst.dataNode.Id && loc.ServerAddress() != sourceVolumeServer {
-			if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false); err != nil {
-				fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err)
+		if preserveServers[loc.Url] {
+			if markErr := markVolumeWritable(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), true, false); markErr != nil {
+				glog.Errorf("mark volume %d as writable on preserved replica %s: %v", vid, loc.Url, markErr)
 			}
-			// reduce volume count? Not really necessary since they are "more" full and will not be a candidate to move to
+		}
+	}
+
+	// Remove old replicas that are NOT needed by the fulfilled replication.
+	// Skip the move destination, the already-deleted source, and any pre-existing
+	// target-tier replicas that were counted toward replication fulfillment.
+	for _, loc := range locations {
+		if loc.Url == dst.dataNode.Id || loc.ServerAddress() == sourceVolumeServer {
+			continue
+		}
+		if preserveServers[loc.Url] {
+			continue
+		}
+		if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false); err != nil {
+			fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err)
 		}
 	}
 	return nil
 }
 
+// restoreSurvivingReplicasWritable marks old replicas writable after a failure,
+// skipping the source that was already deleted by LiveMoveVolume.
+func restoreSurvivingReplicasWritable(commandEnv *CommandEnv, vid needle.VolumeId, locations []wdclient.Location, deletedSource pb.ServerAddress) {
+	for _, loc := range locations {
+		if loc.ServerAddress() == deletedSource {
+			continue
+		}
+		if markErr := markVolumeWritable(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), true, false); markErr != nil {
+			glog.Errorf("mark volume %d as writable on %s: %v", vid, loc.Url, markErr)
+		}
+	}
+}
+
+// ensureReplicationFulfilled creates additional replicas of the volume on the target tier
+// to satisfy the requested replication placement. It re-collects topology after the initial
+// move so it can see the newly placed volume and find suitable destinations for additional copies.
+// It returns a set of server URLs (from the original locations) that host target-tier replicas
+// counted toward fulfillment, so the caller can avoid deleting them during cleanup.
+func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, movedDst location, replicationString string) (preserveServers map[string]bool, err error) {
+	preserveServers = make(map[string]bool)
+	sourceAddress := pb.NewServerAddressFromDataNode(movedDst.dataNode)
+
+	// Wait briefly for the master to receive heartbeats reflecting the move,
+	// then re-collect topology to get the current state.
+	topologyInfo, _, err := collectTopologyInfo(commandEnv, 5*time.Second)
+	if err != nil {
+		return nil, fmt.Errorf("collect topology: %v", err)
+	}
+
+	volumeReplicas, allLocations := collectVolumeReplicaLocations(topologyInfo)
+	allLocations = filterLocationsByDiskType(allLocations, toDiskType)
+	keepDataNodesSorted(allLocations, toDiskType)
+
+	existingReplicas := volumeReplicas[uint32(vid)]
+	if len(existingReplicas) == 0 {
+		return nil, fmt.Errorf("volume %d not found in topology after move", vid)
+	}
+
+	// Build a set of all data nodes that already host this volume (any disk type)
+	// so we don't try to VolumeCopy to a server that already has it.
+	nodesWithVolume := make(map[string]bool)
+	for _, r := range existingReplicas {
+		nodesWithVolume[r.location.dataNode.Id] = true
+	}
+
+	// Determine the target replication: use explicit -toReplication if given,
+	// otherwise read the volume's existing replication setting.
+	var replicaPlacement *super_block.ReplicaPlacement
+	if replicationString != "" {
+		replicaPlacement, err = super_block.NewReplicaPlacementFromString(replicationString)
+		if err != nil {
+			return nil, fmt.Errorf("parse replication %s: %v", replicationString, err)
+		}
+	} else {
+		replicaPlacement, err = super_block.NewReplicaPlacementFromByte(byte(existingReplicas[0].info.ReplicaPlacement))
+		if err != nil {
+			return nil, fmt.Errorf("parse existing replication for volume %d: %v", vid, err)
+		}
+	}
+
+	requiredCopies := replicaPlacement.GetCopyCount()
+	if requiredCopies <= 1 {
+		// No additional replicas needed (e.g., replication "000")
+		return preserveServers, nil
+	}
+
+	// Filter to only replicas on the target disk type (the newly moved one).
+	var targetTierReplicas []*VolumeReplica
+	for _, r := range existingReplicas {
+		if types.ToDiskType(r.info.DiskType) == toDiskType {
+			targetTierReplicas = append(targetTierReplicas, r)
+			// Track pre-existing target-tier replicas so the caller won't delete them.
+			preserveServers[r.location.dataNode.Id] = true
+		}
+	}
+	if len(targetTierReplicas) == 0 {
+		return nil, fmt.Errorf("volume %d not found on target tier %s in topology after move", vid, toDiskType)
+	}
+
+	// Ensure all existing target-tier replicas have the correct replication metadata.
+	// The primary moved replica is already configured in doMoveOneVolume, but there may
+	// be pre-existing replicas on the target tier that need updating.
+	if replicationString != "" {
+		for _, r := range targetTierReplicas {
+			addr := pb.NewServerAddressFromDataNode(r.location.dataNode)
+			if configErr := configureVolumeReplication(commandEnv.option.GrpcDialOption, vid, addr, replicationString); configErr != nil {
+				return nil, fmt.Errorf("volume %d: failed to configure replication on existing replica %s: %v", vid, r.location.dataNode.Id, configErr)
+			}
+		}
+	}
+
+	additionalCopiesNeeded := requiredCopies - len(targetTierReplicas)
+	if additionalCopiesNeeded <= 0 {
+		return preserveServers, nil
+	}
+
+	fmt.Fprintf(writer, "volume %d: creating %d additional replica(s) for replication %s\n", vid, additionalCopiesNeeded, replicaPlacement)
+
+	fn := capacityByFreeVolumeCount(toDiskType)
+	copiesMade := 0
+	for _, candidateDst := range allLocations {
+		if copiesMade >= additionalCopiesNeeded {
+			break
+		}
+		if fn(candidateDst.dataNode) <= 0 {
+			continue
+		}
+		// Skip nodes that already host this volume on any disk type to avoid
+		// VolumeCopy conflicts (e.g., same volume on source tier and target tier).
+		if nodesWithVolume[candidateDst.dataNode.Id] {
+			continue
+		}
+		if !satisfyReplicaPlacement(replicaPlacement, targetTierReplicas, candidateDst) {
+			continue
+		}
+
+		candidateAddress := pb.NewServerAddressFromDataNode(candidateDst.dataNode)
+		fmt.Fprintf(writer, "volume %d: replicating from %s to %s\n", vid, sourceAddress, candidateDst.dataNode.Id)
+
+		if copyErr := replicateVolumeToServer(commandEnv.option.GrpcDialOption, writer, vid, sourceAddress, candidateAddress, toDiskType.ReadableString()); copyErr != nil {
+			return nil, fmt.Errorf("replicate volume %d to %s: %v", vid, candidateDst.dataNode.Id, copyErr)
+		}
+
+		// Configure replication on the new replica if an explicit -toReplication was given.
+		// Without it, VolumeCopy already preserves the source's replication from the super block.
+		if replicationString != "" {
+			if configErr := configureVolumeReplication(commandEnv.option.GrpcDialOption, vid, candidateAddress, replicationString); configErr != nil {
+				return nil, fmt.Errorf("volume %d: failed to configure replication on %s: %v", vid, candidateDst.dataNode.Id, configErr)
+			}
+		}
+
+		// Track the new replica for placement decisions
+		targetTierReplicas = append(targetTierReplicas, &VolumeReplica{
+			location: &candidateDst,
+			info:     targetTierReplicas[0].info,
+		})
+		addVolumeCount(candidateDst.dataNode.DiskInfos[string(toDiskType)], 1)
+		copiesMade++
+	}
+
+	if copiesMade < additionalCopiesNeeded {
+		return nil, fmt.Errorf("could only create %d of %d additional replicas for volume %d (replication %s): not enough eligible destinations", copiesMade, additionalCopiesNeeded, vid, replicaPlacement)
+	}
+
+	fmt.Fprintf(writer, "volume %d: replication %s fulfilled with %d total copies\n", vid, replicaPlacement, requiredCopies)
+	return preserveServers, nil
+}
+
 func collectVolumeIdsForTierChange(topologyInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, sourceTier types.DiskType, collectionPattern string, fullPercentage float64, quietPeriod time.Duration) (vids []needle.VolumeId, err error) {
 
 	quietSeconds := int64(quietPeriod / time.Second)

weed/stats/disk_openbsd.go

@@ -17,7 +17,9 @@ func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
 
 	disk.All = fs.F_blocks * uint64(fs.F_bsize)
 	disk.Free = fs.F_bfree * uint64(fs.F_bsize)
-	calculateDiskRemaining(disk)
+	disk.Used = disk.All - disk.Free
+	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
+	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
 
 	return
 }

weed/stats/disk_solaris.go

@@ -17,7 +17,9 @@ func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
 
 	disk.All = stat.Blocks * uint64(stat.Bsize)
 	disk.Free = stat.Bfree * uint64(stat.Bsize)
-	calculateDiskRemaining(disk)
+	disk.Used = disk.All - disk.Free
+	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
+	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
 
 	return
 }

weed/stats/disk_windows.go

@@ -40,7 +40,9 @@ func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
 
 		return
 	}
-	calculateDiskRemaining(disk)
+	disk.Used = disk.All - disk.Free
+	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
+	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
 
 	return
 }

weed/worker/tasks/ec_balance/ec_balance_task.go

@@ -3,6 +3,8 @@ package ec_balance
 import (
 	"context"
 	"fmt"
+	"math"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -22,7 +24,8 @@ type ECBalanceTask struct {
 	volumeID       uint32
 	collection     string
 	grpcDialOption grpc.DialOption
-	progress       float64
+	progress       uint64 // atomic; stores float64 bits via math.Float64bits
+	reporting      int32  // atomic; re-entry guard to prevent recursive reportProgress calls
 }
 
 // NewECBalanceTask creates a new EC balance task instance
@@ -206,12 +209,16 @@ func (t *ECBalanceTask) EstimateTime(params *worker_pb.TaskParams) time.Duration
 
 // GetProgress returns current progress
 func (t *ECBalanceTask) GetProgress() float64 {
-	return t.progress
+	return math.Float64frombits(atomic.LoadUint64(&t.progress))
 }
 
 // reportProgress updates the stored progress and reports it via the callback
 func (t *ECBalanceTask) reportProgress(progress float64, stage string) {
-	t.progress = progress
+	if !atomic.CompareAndSwapInt32(&t.reporting, 0, 1) {
+		return
+	}
+	defer atomic.StoreInt32(&t.reporting, 0)
+	atomic.StoreUint64(&t.progress, math.Float64bits(progress))
 	t.ReportProgressWithStage(progress, stage)
 	glog.Infof("EC balance volume %d: [%.2f] %s", t.volumeID, progress, stage)
 }