iceberg: detect compaction for spec-evolved tables (#8648 )

iceberg: tighten metadata CAS preconditions
filer: add conditional update preconditions
2026-10-19 13:02:35 +00:00 · 2026-03-16 11:39:22 -07:00 · 2026-03-16 11:39:11 -07:00 · 2026-03-16 11:37:50 -07:00 · 2026-03-16 10:58:08 -07:00 · 2026-03-16 10:47:13 -07:00
339 changed files with 31535 additions and 3777 deletions
@@ -19,7 +19,7 @@ jobs:
      -
        name: Docker meta
        id: docker_meta
-        uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v3
+        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v3
        with:
          images: |
            chrislusf/seaweedfs
@@ -32,7 +32,7 @@ jobs:
            org.opencontainers.image.vendor=Chris Lu
      -
        name: Set up QEMU
-        uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v1
+        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v1
      -
        name: Create BuildKit config
        run: |
@@ -42,28 +42,28 @@ jobs:
          EOF
      -
        name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v1
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v1
        with:
          buildkitd-flags: "--debug"
          buildkitd-config: /tmp/buildkitd.toml
      -
        name: Login to Docker Hub
        if: github.event_name != 'pull_request'
-        uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v1
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
      -
        name: Login to GHCR
        if: github.event_name != 'pull_request'
-        uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v1
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
        with:
          registry: ghcr.io
          username: ${{ secrets.GHCR_USERNAME }}
          password: ${{ secrets.GHCR_TOKEN }}
      -
        name: Build
-        uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v2
+        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v2
        with:
          context: ./docker
          push: ${{ github.event_name != 'pull_request' }}
@@ -126,14 +126,14 @@ jobs:
          echo "seaweedfs_ref=$seaweed" >> "$GITHUB_OUTPUT"

      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@v4

      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

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

      - name: Build and push image
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@v7
        with:
          context: ./docker
          push: ${{ github.event_name != 'pull_request' }}
@@ -86,7 +86,7 @@ jobs:

      - name: Docker meta
        id: docker_meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@v6
        with:
          images: |
            chrislusf/seaweedfs
@@ -98,7 +98,7 @@ jobs:
            org.opencontainers.image.vendor=Chris Lu
      - name: Set up QEMU
        if: matrix.platform != 'amd64'
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@v4
      - name: Create BuildKit config
        run: |
          cat > /tmp/buildkitd.toml <<EOF
@@ -106,25 +106,25 @@ jobs:
            mirrors = ["https://mirror.gcr.io"]
          EOF
      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4
        with:
          buildkitd-flags: "--debug"
          buildkitd-config: /tmp/buildkitd.toml
      - name: Login to Docker Hub
        if: github.event_name != 'pull_request'
-        uses: docker/login-action@v3
+        uses: docker/login-action@v4
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
      - name: Login to GHCR
        if: github.event_name != 'pull_request'
-        uses: docker/login-action@v3
+        uses: docker/login-action@v4
        with:
          registry: ghcr.io
          username: ${{ secrets.GHCR_USERNAME }}
          password: ${{ secrets.GHCR_TOKEN }}
      - name: Build ${{ matrix.platform }} ${{ matrix.variant }}
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@v7
        env:
          DOCKER_BUILDKIT: 1
        with:
@@ -173,19 +173,19 @@ jobs:

      - name: Docker meta
        id: docker_meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@v6
        with:
          images: |
            chrislusf/seaweedfs
            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@v3
+        uses: docker/login-action@v4
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
      - name: Login to GHCR
-        uses: docker/login-action@v3
+        uses: docker/login-action@v4
        with:
          registry: ghcr.io
          username: ${{ secrets.GHCR_USERNAME }}
@@ -21,7 +21,7 @@ jobs:
      -
        name: Docker meta
        id: docker_meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@v6
        with:
          images: |
            chrislusf/seaweedfs
@@ -35,14 +35,14 @@ jobs:
            org.opencontainers.image.vendor=Chris Lu
      -
        name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@v4
      -
        name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4
      -
        name: Login to Docker Hub
        if: github.event_name != 'pull_request'
-        uses: docker/login-action@v3
+        uses: docker/login-action@v4
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
@@ -57,7 +57,7 @@ jobs:
          fi
      -
        name: Build
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@v7
        with:
          context: ./docker
          push: ${{ github.event_name != 'pull_request' }}
@@ -60,16 +60,16 @@ jobs:
            build_args: TAGS=5BytesOffset
            tag_suffix: _large_disk
            
-          # Full tags - amd64 only
+          # Full tags - multi-arch
          - variant: full
-            platforms: linux/amd64
+            platforms: linux/amd64,linux/arm64
            dockerfile: ./docker/Dockerfile.go_build
            build_args: TAGS=elastic,gocdk,rclone,sqlite,tarantool,tikv,ydb
            tag_suffix: _full
-            
-          # Large disk + full tags - amd64 only
+
+          # Large disk + full tags - multi-arch
          - variant: large_disk_full
-            platforms: linux/amd64
+            platforms: linux/amd64,linux/arm64
            dockerfile: ./docker/Dockerfile.go_build
            build_args: TAGS=5BytesOffset,elastic,gocdk,rclone,sqlite,tarantool,tikv,ydb
            tag_suffix: _large_disk_full
@@ -103,7 +103,7 @@ jobs:
      - name: Docker meta
        if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
        id: docker_meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@v6
        with:
          images: |
            chrislusf/seaweedfs
@@ -117,7 +117,7 @@ jobs:
            
      - name: Set up QEMU
        if: (github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant) && contains(matrix.platforms, 'arm')
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@v4
        
      - name: Create BuildKit config
        if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
@@ -129,20 +129,20 @@ jobs:
          
      - name: Set up Docker Buildx
        if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4
        with:
          buildkitd-config: /tmp/buildkitd.toml
          
      - 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@v3
+        uses: docker/login-action@v4
        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@v3
+        uses: docker/login-action@v4
        with:
          registry: ghcr.io
          username: ${{ secrets.GHCR_USERNAME }}
@@ -150,7 +150,7 @@ jobs:
          
      - name: Build and push ${{ matrix.variant }}
        if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@v7
        env:
          DOCKER_BUILDKIT: 1
        with:
@@ -198,14 +198,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@v3
+        uses: docker/login-action@v4
        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@v3
+        uses: docker/login-action@v4
        with:
          registry: ghcr.io
          username: ${{ secrets.GHCR_USERNAME }}
@@ -82,19 +82,19 @@ jobs:
          echo "seaweedfs_ref=$seaweed" >> "$GITHUB_OUTPUT"

      - name: Set up QEMU
-        uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v1
+        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v1

      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v1
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v1

      - name: Login to Docker Hub
-        uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v1
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Build and push image
-        uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v2
+        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v2
        with:
          context: ./docker
          push: true
@@ -26,7 +26,7 @@ jobs:
      - name: Set up Go
        uses: actions/setup-go@v6
        with:
-          go-version: '1.24'
+          go-version-file: 'go.mod'

      - name: Build Telemetry Server
        if: github.event_name == 'workflow_dispatch' && inputs.deploy
@@ -11,4 +11,4 @@ jobs:
      - name: 'Checkout Repository'
        uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8
      - name: 'Dependency Review'
-        uses: actions/dependency-review-action@05fe4576374b728f0c523d6a13d64c25081e0803
+        uses: actions/dependency-review-action@2031cfc080254a8a887f58cffee85186f0e49e48
@@ -23,17 +23,16 @@ jobs:
    runs-on: ubuntu-22.04
    timeout-minutes: 30
    steps:
-    - name: Set up Go 1.x
-      uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
-      with:
-        go-version: ^1.13
-      id: go
-
    - name: Check out code into the Go module directory
-      uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
+      uses: actions/checkout@v6
+
+    - name: Set up Go
+      uses: actions/setup-go@v6
+      with:
+        go-version-file: 'go.mod'

    - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v3
+      uses: docker/setup-buildx-action@v4
      
    - name: Cache Docker layers
      uses: actions/cache@v5
@@ -18,7 +18,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
      id: go

    - name: Check out code into the Go module directory
@@ -28,7 +28,7 @@ jobs:
    - name: Set up Go
      uses: actions/setup-go@v6
      with:
-        go-version: '1.24'
+        go-version-file: 'go.mod'
        
    - name: Build weed binary
      run: |
@@ -22,7 +22,6 @@ permissions:
  contents: read

 env:
-  GO_VERSION: '1.24'
  TEST_TIMEOUT: '45m'

 jobs:
@@ -35,10 +34,10 @@ jobs:
    - name: Checkout code
      uses: actions/checkout@v6
      
-    - name: Set up Go ${{ env.GO_VERSION }}
+    - name: Set up Go
      uses: actions/setup-go@v6
      with:
-        go-version: ${{ env.GO_VERSION }}
+        go-version-file: 'go.mod'
        
    - name: Install FUSE and dependencies
      run: |
@@ -19,13 +19,12 @@ jobs:
    name: Go Vet
    runs-on: ubuntu-latest
    steps:
-    - name: Set up Go 1.x
-      uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
-      with:
-        go-version: ^1.13
-      id: go
    - name: Check out code into the Go module directory
-      uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
+      uses: actions/checkout@v6
+    - name: Set up Go
+      uses: actions/setup-go@v6
+      with:
+        go-version-file: 'go.mod'
    - name: Get dependencies
      run: |
        cd weed; go get -v -t -d ./...
@@ -42,13 +41,12 @@ jobs:
    name: Build
    runs-on: ubuntu-latest
    steps:
-    - name: Set up Go 1.x
-      uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
-      with:
-        go-version: ^1.13
-      id: go
    - name: Check out code into the Go module directory
-      uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
+      uses: actions/checkout@v6
+    - name: Set up Go
+      uses: actions/setup-go@v6
+      with:
+        go-version-file: 'go.mod'
    - name: Build
      run: cd weed; go build -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v .

@@ -56,12 +54,11 @@ jobs:
    name: Test
    runs-on: ubuntu-latest
    steps:
-    - name: Set up Go 1.x
-      uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
-      with:
-        go-version: ^1.13
-      id: go
    - name: Check out code into the Go module directory
-      uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
+      uses: actions/checkout@v6
+    - name: Set up Go
+      uses: actions/setup-go@v6
+      with:
+        go-version-file: 'go.mod'
    - name: Test
      run: cd weed; go test -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v ./...
@@ -116,6 +116,107 @@ jobs:
          echo "✓ COSI driver renders correctly"
          
          echo ""
+          echo "=== Testing long release name: service names match DNS references ==="
+          # Use a release name that, combined with chart name "seaweedfs", exceeds 63 chars.
+          # fullname = "my-very-long-release-name-that-will-cause-truncation-seaweedfs" (65 chars before trunc)
+          LONG_RELEASE="my-very-long-release-name-that-will-cause-truncation"
+
+          # --- Normal mode: master + filer-client services vs helper-produced addresses ---
+          helm template "$LONG_RELEASE" $CHART_DIR \
+            --set s3.enabled=true \
+            --set global.createBuckets[0].name=test > /tmp/longname.yaml
+
+          # Extract Service names from metadata
+          MASTER_SVC=$(awk '/kind: Service/{found=1} found && /^ *name:/{print $2; found=0}' /tmp/longname.yaml \
+            | grep -- '-master$')
+          FILER_CLIENT_SVC=$(awk '/kind: Service/{found=1} found && /^ *name:/{print $2; found=0}' /tmp/longname.yaml \
+            | grep -- '-filer-client$')
+
+          # Extract the hostname from WEED_CLUSTER_SW_MASTER in post-install-bucket-hook
+          MASTER_ADDR=$(grep 'WEED_CLUSTER_SW_MASTER' -A1 /tmp/longname.yaml \
+            | grep 'value:' | head -1 | sed 's/.*value: *"\{0,1\}\([^":]*\).*/\1/')
+          FILER_ADDR=$(grep 'WEED_CLUSTER_SW_FILER' -A1 /tmp/longname.yaml \
+            | grep 'value:' | head -1 | sed 's/.*value: *"\{0,1\}\([^":]*\).*/\1/')
+
+          # Extract the hostname from S3 deployment -filer= argument
+          S3_FILER_HOST=$(grep '\-filer=' /tmp/longname.yaml \
+            | head -1 | sed 's/.*-filer=\([^:]*\).*/\1/')
+
+          # The address helpers produce "<svc>.<namespace>:<port>"; extract just the svc name
+          MASTER_ADDR_SVC=$(echo "$MASTER_ADDR" | cut -d. -f1)
+          FILER_ADDR_SVC=$(echo "$FILER_ADDR" | cut -d. -f1)
+          S3_FILER_SVC=$(echo "$S3_FILER_HOST" | cut -d. -f1)
+
+          echo "  master Service.name:       $MASTER_SVC"
+          echo "  cluster.masterAddress svc:  $MASTER_ADDR_SVC"
+          echo "  filer-client Service.name:  $FILER_CLIENT_SVC"
+          echo "  cluster.filerAddress svc:   $FILER_ADDR_SVC"
+          echo "  S3 -filer= svc:             $S3_FILER_SVC"
+
+          [ "$MASTER_SVC" = "$MASTER_ADDR_SVC" ] || { echo "FAIL: master service name mismatch"; exit 1; }
+          [ "$FILER_CLIENT_SVC" = "$FILER_ADDR_SVC" ] || { echo "FAIL: filer-client service name mismatch"; exit 1; }
+          [ "$FILER_CLIENT_SVC" = "$S3_FILER_SVC" ] || { echo "FAIL: S3 -filer= does not match filer-client service"; exit 1; }
+          echo "✓ Normal mode: service names match DNS references with long release name"
+
+          # --- All-in-one mode: all-in-one service vs both helper addresses ---
+          helm template "$LONG_RELEASE" $CHART_DIR \
+            --set allInOne.enabled=true \
+            --set global.createBuckets[0].name=test > /tmp/longname-aio.yaml
+
+          AIO_SVC=$(awk '/kind: Service/{found=1} found && /^ *name:/{print $2; found=0}' /tmp/longname-aio.yaml \
+            | grep -- '-all-in-one$')
+          AIO_MASTER_ADDR_SVC=$(grep 'WEED_CLUSTER_SW_MASTER' -A1 /tmp/longname-aio.yaml \
+            | grep 'value:' | head -1 | sed 's/.*value: *"\{0,1\}\([^":]*\).*/\1/' | cut -d. -f1)
+          AIO_FILER_ADDR_SVC=$(grep 'WEED_CLUSTER_SW_FILER' -A1 /tmp/longname-aio.yaml \
+            | grep 'value:' | head -1 | sed 's/.*value: *"\{0,1\}\([^":]*\).*/\1/' | cut -d. -f1)
+
+          echo "  all-in-one Service.name:    $AIO_SVC"
+          echo "  cluster.masterAddress svc:  $AIO_MASTER_ADDR_SVC"
+          echo "  cluster.filerAddress svc:   $AIO_FILER_ADDR_SVC"
+
+          [ "$AIO_SVC" = "$AIO_MASTER_ADDR_SVC" ] || { echo "FAIL: all-in-one master address mismatch"; exit 1; }
+          [ "$AIO_SVC" = "$AIO_FILER_ADDR_SVC" ] || { echo "FAIL: all-in-one filer address mismatch"; exit 1; }
+          echo "✓ All-in-one mode: service names match DNS references with long release name"
+
+          echo ""
+          echo "=== Testing security+S3: no blank lines in shell command blocks ==="
+          # Render the three manifests that include seaweedfs.s3.tlsArgs:
+          #   filer-statefulset, s3-deployment, all-in-one-deployment
+          helm template test $CHART_DIR \
+            --set global.enableSecurity=true \
+            --set filer.s3.enabled=true \
+            --set s3.enabled=true > /tmp/security-s3.yaml
+          helm template test $CHART_DIR \
+            --set global.enableSecurity=true \
+            --set allInOne.enabled=true \
+            --set allInOne.s3.enabled=true > /tmp/security-aio.yaml
+
+          pip install pyyaml -q
+          python3 - /tmp/security-s3.yaml /tmp/security-aio.yaml <<'PYEOF'
+          import yaml, sys
+          errors = []
+          for path in sys.argv[1:]:
+              with open(path) as f:
+                  docs = list(yaml.safe_load_all(f))
+              for doc in docs:
+                  if not doc or doc.get("kind") not in ("Deployment", "StatefulSet"):
+                      continue
+                  name = doc["metadata"]["name"]
+                  for c in doc["spec"]["template"]["spec"].get("containers", []):
+                      cmd = c.get("command", [])
+                      if len(cmd) >= 3 and cmd[0] == "/bin/sh" and cmd[1] == "-ec":
+                          script = cmd[2]
+                          for i, line in enumerate(script.splitlines(), 1):
+                              if line.strip() == "":
+                                  errors.append(f"{path}: {name}/{c['name']} has blank line at script line {i}")
+          if errors:
+              for e in errors:
+                  print(f"FAIL: {e}", file=sys.stderr)
+              print("Rendered with: global.enableSecurity=true, filer.s3.enabled=true, s3.enabled=true, allInOne.enabled=true", file=sys.stderr)
+              sys.exit(1)
+          print("✓ No blank lines in security+S3 command blocks")
+          PYEOF
+
          echo "✅ All template rendering tests passed!"

      - name: Create kind cluster
@@ -26,14 +26,14 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
        cache: true
        cache-dependency-path: |
          **/go.sum
      id: go

    - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v3
+      uses: docker/setup-buildx-action@v4

    - name: Install dependencies
      run: |
@@ -37,7 +37,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
      id: go

    - name: Check out code
@@ -82,7 +82,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
      id: go

    - name: Check out code
@@ -132,7 +132,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
        cache: true
        cache-dependency-path: |
          **/go.sum
@@ -311,7 +311,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
        cache: true
        cache-dependency-path: |
          **/go.sum
@@ -473,7 +473,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
        cache: true
        cache-dependency-path: |
          **/go.sum
@@ -631,7 +631,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
        cache: true
        cache-dependency-path: |
          **/go.sum
@@ -789,7 +789,7 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
      id: go

    - name: Check out code
@@ -30,7 +30,6 @@ permissions:
  contents: read

 env:
-  GO_VERSION: '1.24'
  TEST_TIMEOUT: '10m'

 jobs:
@@ -43,10 +42,10 @@ jobs:
    - name: Checkout code
      uses: actions/checkout@v6
      
-    - name: Set up Go ${{ env.GO_VERSION }}
+    - name: Set up Go
      uses: actions/setup-go@v6
      with:
-        go-version: ${{ env.GO_VERSION }}
+        go-version-file: 'go.mod'
        
    - name: Build SeaweedFS
      run: |
@@ -25,14 +25,14 @@ jobs:
    - name: Set up Go 1.x
      uses: actions/setup-go@v6
      with:
-        go-version: ^1.24
+        go-version: ^1.25
      id: go

    - name: Check out code
      uses: actions/checkout@v6

    - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v3
+      uses: docker/setup-buildx-action@v4

    - name: Cache Docker layers
      uses: actions/cache@v5
@@ -5,6 +5,8 @@ on:
    paths:
      - 'weed/iam/**'
      - 'weed/s3api/**'
+      - 'weed/credential/**'
+      - 'weed/pb/**'
      - 'test/s3/iam/**'
      - '.github/workflows/s3-iam-tests.yml'
  push:
@@ -12,6 +14,8 @@ on:
    paths:
      - 'weed/iam/**'
      - 'weed/s3api/**'
+      - 'weed/credential/**'
+      - 'weed/pb/**'
      - 'test/s3/iam/**'
      - '.github/workflows/s3-iam-tests.yml'
  
@@ -80,7 +84,7 @@ jobs:
    timeout-minutes: 25
    strategy:
      matrix:
-        test-type: ["basic", "advanced", "policy-enforcement"]
+        test-type: ["basic", "advanced", "policy-enforcement", "group"]
        
    steps:
      - name: Check out code
@@ -117,7 +121,7 @@ jobs:
            "basic")
              echo "Running basic IAM functionality tests..."
              make clean setup start-services wait-for-services
-              go test -v -timeout 15m -run "TestS3IAMAuthentication|TestS3IAMBasicWorkflow|TestS3IAMTokenValidation|TestIAM" ./...
+              go test -v -timeout 15m -run "TestS3IAMAuthentication|TestS3IAMBasicWorkflow|TestS3IAMTokenValidation|TestIAMUserManagement|TestIAMAccessKeyManagement|TestIAMPolicyManagement" ./...
              ;;
            "advanced")
              echo "Running advanced IAM feature tests..."
@@ -129,6 +133,11 @@ jobs:
              make clean setup start-services wait-for-services
              go test -v -timeout 15m -run "TestS3IAMPolicyEnforcement|TestS3IAMBucketPolicy|TestS3IAMContextual" ./...
              ;;
+            "group")
+              echo "Running IAM group management tests..."
+              make clean setup start-services wait-for-services
+              go test -v -timeout 15m -run "TestIAMGroup" ./...
+              ;;
            *)
              echo "Unknown test type: ${{ matrix.test-type }}"
              exit 1
@@ -41,7 +41,7 @@ jobs:
      - name: Set up Go
        uses: actions/setup-go@v6
        with:
-          go-version: ^1.24
+          go-version: ^1.25
          cache: true
      
      - name: Set up Python ${{ matrix.python-version }}
@@ -148,7 +148,7 @@ jobs:
      - name: Set up Go
        uses: actions/setup-go@v6
        with:
-          go-version: ^1.24
+          go-version: ^1.25
          cache: true
      
      - name: Run Go unit tests
@@ -29,7 +29,7 @@ jobs:
        id: go

      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Build SeaweedFS binary for Linux
        run: |
@@ -33,7 +33,7 @@ jobs:
        id: go

      - name: Set up Docker
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Install SeaweedFS
        run: |
@@ -144,7 +144,7 @@ jobs:
        id: go

      - name: Set up Docker
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Pre-pull Trino image
        run: docker pull trinodb/trino:479
@@ -271,7 +271,7 @@ jobs:
        id: go

      - name: Set up Docker
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Pre-pull Spark image
        run: docker pull apache/spark:3.5.1
@@ -337,7 +337,7 @@ jobs:
        id: go

      - name: Set up Docker
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Pre-pull RisingWave image
        run: |
@@ -405,7 +405,7 @@ jobs:
        id: go

      - name: Set up Docker
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Pre-pull Python image
        run: docker pull python:3
@@ -471,7 +471,7 @@ jobs:
        id: go

      - name: Set up Docker
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4

      - name: Pre-pull Python image
        run: docker pull python:3
@@ -24,7 +24,6 @@ permissions:
  contents: read

 env:
-  GO_VERSION: '1.24'
  TEST_TIMEOUT: '15m'

 jobs:
@@ -37,10 +36,10 @@ jobs:
    - name: Checkout code
      uses: actions/checkout@v6
      
-    - name: Set up Go ${{ env.GO_VERSION }}
+    - name: Set up Go
      uses: actions/setup-go@v6
      with:
-        go-version: ${{ env.GO_VERSION }}
+        go-version-file: 'go.mod'
        
    - name: Install dependencies
      run: |
@@ -43,7 +43,7 @@ jobs:
    - name: Set up Go
      uses: actions/setup-go@v6
      with:
-        go-version: '1.24'
+        go-version-file: 'go.mod'

    - name: Build SeaweedFS binary
      run: |
@@ -24,7 +24,7 @@ jobs:

      - uses: actions/setup-go@v6
        with:
-          go-version: ^1.24
+          go-version: ^1.25

      - name: Build SeaweedFS
        run: |
@@ -28,7 +28,6 @@ permissions:
  contents: read

 env:
-  GO_VERSION: '1.24'
  TEST_TIMEOUT: '30m'

 jobs:
@@ -46,10 +45,10 @@ jobs:
      - name: Checkout code
        uses: actions/checkout@v6

-      - name: Set up Go ${{ env.GO_VERSION }}
+      - name: Set up Go
        uses: actions/setup-go@v6
        with:
-          go-version: ${{ env.GO_VERSION }}
+          go-version-file: 'go.mod'

      - name: Build SeaweedFS binary
        run: |
@@ -1,4 +1,4 @@
-FROM golang:1.24 AS builder
+FROM golang:1.25 AS builder

 RUN apt-get update && \
    apt-get install -y build-essential wget ca-certificates && \
@@ -1,4 +1,4 @@
-FROM golang:1.24-alpine as builder
+FROM golang:1.25-alpine AS builder
 RUN apk add git g++ fuse
 RUN mkdir -p /go/src/github.com/seaweedfs/
 ARG BRANCH=${BRANCH:-master}
@@ -1,4 +1,4 @@
-FROM golang:1.24 AS builder
+FROM golang:1.25 AS builder

 RUN apt-get update
 RUN apt-get install -y build-essential libsnappy-dev zlib1g-dev libbz2-dev libgflags-dev liblz4-dev libzstd-dev
@@ -1,4 +1,4 @@
-FROM golang:1.24 AS builder
+FROM golang:1.25 AS builder

 RUN apt-get update
 RUN apt-get install -y build-essential libsnappy-dev zlib1g-dev libbz2-dev libgflags-dev liblz4-dev libzstd-dev
@@ -1,6 +1,6 @@
 module github.com/seaweedfs/seaweedfs

-go 1.24.9
+go 1.25.0

 require (
 	cloud.google.com/go v0.123.0 // indirect
@@ -63,7 +63,7 @@ require (
 	github.com/prometheus/client_golang v1.23.2
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.67.2 // indirect
-	github.com/prometheus/procfs v0.19.2
+	github.com/prometheus/procfs v0.20.1
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/seaweedfs/goexif v1.0.3
@@ -87,7 +87,7 @@ require (
 	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.7
-	go.mongodb.org/mongo-driver v1.17.6
+	go.mongodb.org/mongo-driver v1.17.9
 	go.opencensus.io v0.24.0 // indirect
 	gocloud.dev v0.45.0
 	gocloud.dev/pubsub/natspubsub v0.45.0
@@ -96,8 +96,8 @@ require (
 	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546
 	golang.org/x/image v0.36.0
 	golang.org/x/net v0.49.0
-	golang.org/x/oauth2 v0.35.0
-	golang.org/x/sys v0.41.0
+	golang.org/x/oauth2 v0.36.0
+	golang.org/x/sys v0.42.0
 	golang.org/x/text v0.34.0 // indirect
 	golang.org/x/tools v0.41.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
@@ -124,13 +124,13 @@ require (
 	github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3
 	github.com/arangodb/go-driver v1.6.9
 	github.com/armon/go-metrics v0.4.1
-	github.com/aws/aws-sdk-go-v2 v1.41.1
+	github.com/aws/aws-sdk-go-v2 v1.41.4
 	github.com/aws/aws-sdk-go-v2/config v1.32.7
-	github.com/aws/aws-sdk-go-v2/credentials v1.19.7
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.12
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.95.0
 	github.com/cognusion/imaging v1.0.2
 	github.com/fluent/fluent-logger-golang v1.10.1
-	github.com/getsentry/sentry-go v0.42.0
+	github.com/getsentry/sentry-go v0.43.0
 	github.com/go-ldap/ldap/v3 v3.4.12
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/flatbuffers/go v0.0.0-20230108230133-3b8644d32c50
@@ -151,7 +151,7 @@ require (
 	github.com/schollz/progressbar/v3 v3.19.0
 	github.com/seaweedfs/go-fuse/v2 v2.9.1
 	github.com/shirou/gopsutil/v4 v4.26.2
-	github.com/tarantool/go-tarantool/v2 v2.4.1
+	github.com/tarantool/go-tarantool/v2 v2.4.2
 	github.com/testcontainers/testcontainers-go v0.39.0
 	github.com/tikv/client-go/v2 v2.0.7
 	github.com/xeipuuv/gojsonschema v1.2.0
@@ -183,7 +183,7 @@ require (
 	github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
 	github.com/apache/arrow-go/v18 v18.4.1 // indirect
 	github.com/apache/thrift v0.22.0 // indirect
-	github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/signin v1.0.8 // indirect
 	github.com/bahlo/generic-list-go v0.2.0 // indirect
 	github.com/bazelbuild/rules_go v0.46.0 // indirect
 	github.com/biogo/store v0.0.0-20201120204734-aad293a2328f // indirect
@@ -326,22 +326,22 @@ require (
 	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/feature/ec2/imds v1.18.17 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.20 // 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.17 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.20 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.20 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.16 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // 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.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.20 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.16 // 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.9 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6
-	github.com/aws/smithy-go v1.24.0
+	github.com/aws/aws-sdk-go-v2/service/sso v1.30.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.41.9
+	github.com/aws/smithy-go v1.24.2
 	github.com/boltdb/bolt v1.3.1 // indirect
 	github.com/bradenaw/juniper v0.15.3 // indirect
 	github.com/bradfitz/iter v0.0.0-20191230175014-e8f45d346db8 // indirect
@@ -443,7 +443,7 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pengsrc/go-shared v0.2.1-0.20190131101655-1999055a4a14 // indirect
 	github.com/philhofer/fwd v1.2.0 // indirect
-	github.com/pierrec/lz4/v4 v4.1.25
+	github.com/pierrec/lz4/v4 v4.1.26
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
 	github.com/pingcap/kvproto v0.0.0-20230403051650-e166ae588106 // indirect
@@ -1,6 +1,6 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "4.15"
+appVersion: "4.17"
 # Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
-version: 4.15.0
+version: 4.17.0
@@ -165,8 +165,9 @@ admin:
  enabled: true
  port: 23646
  grpcPort: 33646  # For worker connections
-  adminUser: "admin"
-  adminPassword: "your-secure-password"  # Leave empty to disable auth
+  secret:
+    adminUser: "admin"
+    adminPassword: "your-secure-password"  # Leave empty to disable auth
  
  # Optional: persist admin data
  data:
@@ -191,6 +192,8 @@ If `adminPassword` is set, the admin interface requires authentication:

 If `adminPassword` is empty or not set, the admin interface runs without authentication (not recommended for production).

+As an alternative, a kubernetes Secret can be used (`admin.secret.existingSecret`).
+
 ### Admin Data Persistence

 The admin component can store configuration and maintenance data. You can configure storage in several ways:
@@ -1539,7 +1539,7 @@
      "pluginVersion": "10.3.1",
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total{namespace=\"$NAMESPACE\"}[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total{namespace=\"$NAMESPACE\"}[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -1584,7 +1584,7 @@
      "pluginVersion": "10.3.1",
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total{namespace=\"$NAMESPACE\"}[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total{namespace=\"$NAMESPACE\"}[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -1679,7 +1679,7 @@
      "pluginVersion": "10.3.1",
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_request_total{namespace=\"$NAMESPACE\"}[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_request_total{namespace=\"$NAMESPACE\"}[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -23,8 +23,15 @@
 #
 # Adjust storageClass and sizes to match your cluster's available StorageClasses.
 # On OpenShift you can discover them with: oc get storageclass
-
+global:
+  enableReplication: true
+  #  replication type is XYZ:
+  # X number of replica in other data centers
+  # Y number of replica in other racks in the same data center
+  # Z number of replica in other servers in the same rack
+  replicationPlacement: "000" # no data replica
 master:
+  replicas: 1
  data:
    type: "persistentVolumeClaim"
    size: "10Gi"
@@ -49,6 +56,7 @@ master:
      type: RuntimeDefault

 volume:
+  replicas: 1
  dataDirs:
    - name: data1
      type: "persistentVolumeClaim"
@@ -75,6 +83,7 @@ volume:
      type: RuntimeDefault

 filer:
+  replicas: 1
  data:
    type: "persistentVolumeClaim"
    size: "25Gi"
@@ -93,12 +93,12 @@ spec:
          {{- $adminSecretName := .Values.admin.secret.existingSecret | default (printf "%s-admin-secret" (include "seaweedfs.fullname" .)) }}
          env:
            {{- if $adminAuthEnabled }}
-            - name: SEAWEEDFS_ADMIN_USER
+            - name: WEED_ADMIN_USER
              valueFrom:
                secretKeyRef:
                  name: {{ $adminSecretName }}
                  key: {{ if .Values.admin.secret.existingSecret }}{{ .Values.admin.secret.userKey }}{{ else }}adminUser{{ end }}
-            - name: SEAWEEDFS_ADMIN_PASSWORD
+            - name: WEED_ADMIN_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: {{ $adminSecretName }}
@@ -153,10 +153,6 @@ spec:
              {{- else if .Values.admin.dataDir }}
              -dataDir={{ .Values.admin.dataDir }} \
              {{- end }}
-              {{- if $adminAuthEnabled }}
-              -adminUser="${SEAWEEDFS_ADMIN_USER}" \
-              -adminPassword="${SEAWEEDFS_ADMIN_PASSWORD}" \
-              {{- end }}
              {{- if .Values.admin.masters }}
              -masters={{ .Values.admin.masters }}{{- if .Values.admin.extraArgs }} \{{ end }}
              {{- else if .Values.global.masterServer }}
@@ -243,8 +243,7 @@ spec:
              {{- if $httpsPort }}
              -s3.port.https={{ $httpsPort }} \
              {{- end }}
-              -s3.cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-              -s3.key.file=/usr/local/share/ca-certificates/client/tls.key \
+              {{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }}
              {{- end }}
              {{- if or .Values.allInOne.s3.enableAuth .Values.s3.enableAuth .Values.filer.s3.enableAuth }}
              -s3.config=/etc/sw/s3/seaweedfs_s3_config \
@@ -346,6 +345,9 @@ spec:
            - name: client-cert
              mountPath: /usr/local/share/ca-certificates/client/
              readOnly: true
+            {{- if .Values.allInOne.s3.enabled }}
+            {{- include "seaweedfs.s3.tlsVolumeMount" . | nindent 12 }}
+            {{- end }}
            {{- end }}
            {{ tpl .Values.allInOne.extraVolumeMounts . | nindent 12 }}
          ports:
@@ -473,6 +475,9 @@ spec:
        - name: client-cert
          secret:
            secretName: {{ include "seaweedfs.fullname" . }}-client-cert
+        {{- if .Values.allInOne.s3.enabled }}
+        {{- include "seaweedfs.s3.tlsVolume" . | nindent 8 }}
+        {{- end }}
        {{- end }}
        {{ tpl .Values.allInOne.extraVolumes . | nindent 8 }}
      {{- if .Values.allInOne.nodeSelector }}
@@ -2,7 +2,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-all-in-one" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "all-in-one") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -17,6 +17,9 @@ metadata:
 spec:
  type: {{ .Values.allInOne.service.type | default "ClusterIP" }}
  internalTrafficPolicy: {{ .Values.allInOne.service.internalTrafficPolicy | default "Cluster" }}
+  {{- if and (semverCompare ">=1.31-0" .Capabilities.KubeVersion.GitVersion) .Values.allInOne.s3.trafficDistribution }}
+  trafficDistribution: {{ include "seaweedfs.trafficDistribution" (dict "value" .Values.allInOne.s3.trafficDistribution "Capabilities" .Capabilities) }}
+  {{- end }}
  ports:
  # Master ports
  - name: "swfs-master"
@@ -2,7 +2,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-filer-client" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "filer-client") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -4,7 +4,7 @@ kind: Service
 metadata:
  annotations:
    service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
-  name: {{ printf "%s-filer" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "filer") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -200,8 +200,7 @@ spec:
              {{- if .Values.filer.s3.httpsPort }}
              -s3.port.https={{ .Values.filer.s3.httpsPort }} \
              {{- end }}
-              -s3.cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-              -s3.key.file=/usr/local/share/ca-certificates/client/tls.key \
+              {{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }}
              {{- end }}
              {{- if .Values.filer.s3.enableAuth }}
              -s3.config=/etc/sw/seaweedfs_s3_config \
@@ -254,6 +253,9 @@ spec:
            - name: client-cert
              readOnly: true
              mountPath: /usr/local/share/ca-certificates/client
+            {{- if .Values.filer.s3.enabled }}
+            {{- include "seaweedfs.s3.tlsVolumeMount" . | nindent 12 }}
+            {{- end }}
            {{- end }}
            {{ tpl .Values.filer.extraVolumeMounts . | nindent 12 | trim }}
          ports:
@@ -384,6 +386,9 @@ spec:
        - name: client-cert
          secret:
            secretName: {{ include "seaweedfs.fullname" . }}-client-cert
+        {{- if .Values.filer.s3.enabled }}
+        {{- include "seaweedfs.s3.tlsVolume" . | nindent 8 }}
+        {{- end }}
        {{- end }}
        {{ tpl .Values.filer.extraVolumes . | indent 8 | trim }}
      {{- if .Values.filer.nodeSelector }}
@@ -2,7 +2,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-master" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "master") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -127,8 +127,7 @@ spec:
              {{- if .Values.s3.httpsPort }}
              -port.https={{ .Values.s3.httpsPort }} \
              {{- end }}
-              -cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-              -key.file=/usr/local/share/ca-certificates/client/tls.key \
+              {{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "") | nindent 14 }}
              {{- end }}
              {{- if .Values.s3.domainName }}
              -domainName={{ .Values.s3.domainName }} \
@@ -176,6 +175,7 @@ spec:
            - name: client-cert
              readOnly: true
              mountPath: /usr/local/share/ca-certificates/client/
+            {{- include "seaweedfs.s3.tlsVolumeMount" . | nindent 12 }}
            {{- end }}
            {{ tpl .Values.s3.extraVolumeMounts . | nindent 12 | trim }}
          ports:
@@ -267,6 +267,7 @@ spec:
        - name: client-cert
          secret:
            secretName: {{ include "seaweedfs.fullname" . }}-client-cert
+        {{- include "seaweedfs.s3.tlsVolume" . | nindent 8 }}
        {{- end }}
        {{ tpl .Values.s3.extraVolumes . | indent 8 | trim }}
      {{- if .Values.s3.nodeSelector }}
@@ -2,7 +2,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-s3" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "s3") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -16,8 +16,9 @@ metadata:
 {{- end }}
 spec:
  internalTrafficPolicy: {{ .Values.s3.internalTrafficPolicy | default "Cluster" }}
-  {{- if and (semverCompare ">=1.31-0" .Capabilities.KubeVersion.GitVersion) (or .Values.s3.trafficDistribution .Values.filer.s3.trafficDistribution) }}
-  trafficDistribution: {{ include "seaweedfs.trafficDistribution" . }}
+  {{- $td := .Values.s3.trafficDistribution | default .Values.filer.s3.trafficDistribution }}
+  {{- if and (semverCompare ">=1.31-0" .Capabilities.KubeVersion.GitVersion) $td }}
+  trafficDistribution: {{ include "seaweedfs.trafficDistribution" (dict "value" $td "Capabilities" .Capabilities) }}
  {{- end }}
  ports:
  - name: "swfs-s3"
@@ -49,7 +49,7 @@ spec:
      {{- end }}
      {{- if .Values.sftp.topologySpreadConstraints }}
      topologySpreadConstraints:
-        {{ tpl .Values.sftp.topologySpreadConstraint . | nindent 8 | trim }}
+        {{ tpl .Values.sftp.topologySpreadConstraints . | nindent 8 | trim }}
      {{- end }}
      {{- if .Values.sftp.tolerations }}
      tolerations:
@@ -2,7 +2,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-sftp" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "sftp") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -287,11 +287,8 @@ Compute the master service address to be used in cluster env vars.
 If allInOne is enabled, point to the all-in-one service; otherwise, point to the master service.
 */}}
 {{- define "seaweedfs.cluster.masterAddress" -}}
-{{- $serviceNameSuffix := "-master" -}}
-{{- if .Values.allInOne.enabled -}}
-{{-   $serviceNameSuffix = "-all-in-one" -}}
-{{- end -}}
-{{- printf "%s.%s:%d" (printf "%s%s" (include "seaweedfs.fullname" .) $serviceNameSuffix | trunc 63 | trimSuffix "-") .Release.Namespace (int .Values.master.port) -}}
+{{- $component := ternary "all-in-one" "master" .Values.allInOne.enabled -}}
+{{- printf "%s.%s:%d" (include "seaweedfs.componentName" (list . $component)) .Release.Namespace (int .Values.master.port) -}}
 {{- end -}}

 {{/*
@@ -299,11 +296,8 @@ Compute the filer service address to be used in cluster env vars.
 If allInOne is enabled, point to the all-in-one service; otherwise, point to the filer-client service.
 */}}
 {{- define "seaweedfs.cluster.filerAddress" -}}
-{{- $serviceNameSuffix := "-filer-client" -}}
-{{- if .Values.allInOne.enabled -}}
-{{-   $serviceNameSuffix = "-all-in-one" -}}
-{{- end -}}
-{{- printf "%s.%s:%d" (printf "%s%s" (include "seaweedfs.fullname" .) $serviceNameSuffix | trunc 63 | trimSuffix "-") .Release.Namespace (int .Values.filer.port) -}}
+{{- $component := ternary "all-in-one" "filer-client" .Values.allInOne.enabled -}}
+{{- printf "%s.%s:%d" (include "seaweedfs.componentName" (list . $component)) .Release.Namespace (int .Values.filer.port) -}}
 {{- end -}}

 {{/*
@@ -338,11 +332,41 @@ Create the name of the service account to use
 {{- .Values.global.serviceAccountName | default "seaweedfs" -}}
 {{- end -}}

-{{/* Generate a compatible trafficDistribution value due to "PreferClose" fast deprecation in k8s v1.35 */}}
+{{/* S3 TLS cert/key arguments, using custom secret if s3.tlsSecret is set */}}
+{{- define "seaweedfs.s3.tlsArgs" -}}
+{{- $prefix := .prefix -}}
+{{- $root := .root -}}
+{{- if $root.Values.s3.tlsSecret -}}
+-{{ $prefix }}cert.file=/usr/local/share/ca-certificates/s3/tls.crt \
+-{{ $prefix }}key.file=/usr/local/share/ca-certificates/s3/tls.key \
+{{- else -}}
+-{{ $prefix }}cert.file=/usr/local/share/ca-certificates/client/tls.crt \
+-{{ $prefix }}key.file=/usr/local/share/ca-certificates/client/tls.key \
+{{- end -}}
+{{- end -}}
+
+{{/* S3 custom TLS volume mount */}}
+{{- define "seaweedfs.s3.tlsVolumeMount" -}}
+{{- if .Values.s3.tlsSecret }}
+- name: s3-tls-cert
+  readOnly: true
+  mountPath: /usr/local/share/ca-certificates/s3/
+{{- end }}
+{{- end -}}
+
+{{/* S3 custom TLS volume */}}
+{{- define "seaweedfs.s3.tlsVolume" -}}
+{{- if .Values.s3.tlsSecret }}
+- name: s3-tls-cert
+  secret:
+    secretName: {{ .Values.s3.tlsSecret }}
+{{- end }}
+{{- end -}}
+
+{{/* Generate a compatible trafficDistribution value due to "PreferClose" fast deprecation in k8s v1.35.
+     Accepts a dict with "value" (the trafficDistribution string) and "Capabilities". */}}
 {{- define "seaweedfs.trafficDistribution" -}}
-{{- if .Values.s3.trafficDistribution -}}
-{{- and (eq .Values.s3.trafficDistribution "PreferClose") (semverCompare ">=1.35-0" .Capabilities.KubeVersion.GitVersion) | ternary "PreferSameZone" .Values.s3.trafficDistribution -}}
-{{- else if .Values.filer.s3.trafficDistribution -}}
-{{- and (eq .Values.filer.s3.trafficDistribution "PreferClose") (semverCompare ">=1.35-0" .Capabilities.KubeVersion.GitVersion) | ternary "PreferSameZone" .Values.filer.s3.trafficDistribution -}}
+{{- if .value -}}
+{{- and (eq .value "PreferClose") (semverCompare ">=1.35-0" .Capabilities.KubeVersion.GitVersion) | ternary "PreferSameZone" .value -}}
 {{- end -}}
 {{- end -}}
@@ -8,7 +8,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-%s" (include "seaweedfs.fullname" $) $volumeName | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list $ $volumeName) }}
  namespace: {{ $.Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" $ }}
@@ -2,7 +2,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ printf "%s-worker" (include "seaweedfs.fullname" .) | trunc 63 | trimSuffix "-" }}
+  name: {{ include "seaweedfs.componentName" (list . "worker") }}
  namespace: {{ .Release.Namespace }}
  labels:
    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
@@ -919,6 +919,13 @@ s3:
  port: 8333
  # add additional https port
  httpsPort: 0
+  # Use a custom TLS certificate secret for the S3 HTTPS endpoint.
+  # When set, this Kubernetes Secret (must contain tls.crt and tls.key) is used
+  # instead of the internal self-signed client certificate generated by cert-manager.
+  # This allows using a publicly trusted certificate (e.g., from Let's Encrypt)
+  # so that S3 clients don't need to trust the internal CA.
+  # Requires global.enableSecurity to be true.
+  tlsSecret: null
  metricsPort: 9327
  # Iceberg catalog REST port (Apache Iceberg REST Catalog API)
  # Set to a port number to enable, or 0/null to disable
@@ -1453,6 +1460,7 @@ allInOne:
    # The s3-secret.yaml template only reads from .Values.s3.credentials.
    # See: s3.credentials.admin.accessKey, s3.credentials.read.accessKey
    auditLogConfig: null  # S3 audit log configuration (null inherits from s3.auditLogConfig)
+    trafficDistribution: null  # Service traffic distribution (e.g., "PreferClose"); auto-converts to "PreferSameZone" on k8s >=1.35
    # You may specify buckets to be created during the install process.
    # Buckets may be exposed publicly by setting `anonymousRead` to `true`
    # ttl format: [1-255][m|h|d|w|M|y] (e.g., 7d)
@@ -100,10 +100,12 @@ message ListEntriesRequest {
    string startFromFileName = 3;
    bool inclusiveStartFrom = 4;
    uint32 limit = 5;
+    int64 snapshot_ts_ns = 6;
 }

 message ListEntriesResponse {
    Entry entry = 1;
+    int64 snapshot_ts_ns = 2;
 }

 message RemoteEntry {
@@ -203,6 +205,7 @@ message CreateEntryRequest {

 message CreateEntryResponse {
    string error = 1;
+    SubscribeMetadataResponse metadata_event = 2;
 }

 message UpdateEntryRequest {
@@ -210,8 +213,10 @@ message UpdateEntryRequest {
    Entry entry = 2;
    bool is_from_other_cluster = 3;
    repeated int32 signatures = 4;
+    map<string, bytes> expected_extended = 5;
 }
 message UpdateEntryResponse {
+    SubscribeMetadataResponse metadata_event = 1;
 }

 message AppendToEntryRequest {
@@ -236,6 +241,7 @@ message DeleteEntryRequest {

 message DeleteEntryResponse {
    string error = 1;
+    SubscribeMetadataResponse metadata_event = 2;
 }

 message AtomicRenameEntryRequest {
@@ -469,6 +475,7 @@ message CacheRemoteObjectToLocalClusterRequest {
 }
 message CacheRemoteObjectToLocalClusterResponse {
    Entry entry = 1;
+    SubscribeMetadataResponse metadata_event = 2;
 }

 /////////////////////////
@@ -465,7 +465,7 @@
          "bars": false,
          "dashLength": 10,
          "dashes": false,
-          "datasource": "Prometheus",
+          "datasource": "${DS_PROMETHEUS}",
          "fill": 1,
          "gridPos": {
            "h": 6,
@@ -547,7 +547,7 @@
          "bars": false,
          "dashLength": 10,
          "dashes": false,
-          "datasource": "Prometheus",
+          "datasource": "${DS_PROMETHEUS}",
          "fill": 1,
          "gridPos": {
            "h": 6,
@@ -649,7 +649,7 @@
          "bars": false,
          "dashLength": 10,
          "dashes": false,
-          "datasource": "Prometheus",
+          "datasource": "${DS_PROMETHEUS}",
          "fill": 1,
          "gridPos": {
            "h": 6,
@@ -731,7 +731,7 @@
          "bars": false,
          "dashLength": 10,
          "dashes": false,
-          "datasource": "Prometheus",
+          "datasource": "${DS_PROMETHEUS}",
          "fill": 1,
          "gridPos": {
            "h": 6,
@@ -1182,7 +1182,7 @@
      "steppedLine": false,
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -1275,7 +1275,7 @@
      "steppedLine": false,
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -1368,7 +1368,7 @@
      "steppedLine": false,
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_request_total[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_request_total[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -3309,4 +3309,4 @@
  "title": "SeaweedFS",
  "uid": "nh02dOVnz",
  "version": 2
-}
+}
@@ -754,7 +754,7 @@
          "steppedLine": false,
          "targets": [
            {
-              "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total[$__interval])) by (bucket)",
+              "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total[$__rate_interval])) by (bucket)",
              "format": "time_series",
              "hide": false,
              "intervalFactor": 2,
@@ -835,7 +835,7 @@
          "steppedLine": false,
          "targets": [
            {
-              "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total[$__interval])) by (bucket)",
+              "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total[$__rate_interval])) by (bucket)",
              "format": "time_series",
              "hide": false,
              "intervalFactor": 2,
@@ -914,7 +914,7 @@
          "steppedLine": false,
          "targets": [
            {
-              "expr": "sum(rate(SeaweedFS_s3_request_total[$__interval])) by (bucket)",
+              "expr": "sum(rate(SeaweedFS_s3_request_total[$__rate_interval])) by (bucket)",
              "format": "time_series",
              "hide": false,
              "intervalFactor": 2,
@@ -310,7 +310,7 @@
      "steppedLine": false,
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total{namespace=\"$namespace\",service=~\"$service-api\"}[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_received_bytes_total{namespace=\"$namespace\",service=~\"$service-api\"}[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -403,7 +403,7 @@
      "steppedLine": false,
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total{namespace=\"$namespace\",service=~\"$service-api\"}[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_bucket_traffic_sent_bytes_total{namespace=\"$namespace\",service=~\"$service-api\"}[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -496,7 +496,7 @@
      "steppedLine": false,
      "targets": [
        {
-          "expr": "sum(rate(SeaweedFS_s3_request_total{namespace=\"$namespace\",service=~\"$service-api\"}[$__interval])) by (bucket)",
+          "expr": "sum(rate(SeaweedFS_s3_request_total{namespace=\"$namespace\",service=~\"$service-api\"}[$__rate_interval])) by (bucket)",
          "format": "time_series",
          "hide": false,
          "intervalFactor": 2,
@@ -1,5 +1,5 @@
 # Multi-stage build for Go Sidecar
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Install build dependencies
 RUN apk add --no-cache git ca-certificates tzdata
@@ -1,5 +1,5 @@
 # Multi-stage build for Test Client
-FROM golang:1.23-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Install build dependencies
 RUN apk add --no-cache git ca-certificates tzdata
@@ -194,28 +194,6 @@ func TestEcEndToEnd(t *testing.T) {
 	// 1. Configure plugin job types for fast EC detection/execution.
 	t.Log("Configuring plugin job types via API...")

-	schedulerConfig := map[string]interface{}{
-		"idle_sleep_seconds": 1,
-	}
-	jsonBody, err := json.Marshal(schedulerConfig)
-	if err != nil {
-		t.Fatalf("Failed to marshal scheduler config: %v", err)
-	}
-	req, err := http.NewRequest("PUT", AdminUrl+"/api/plugin/scheduler-config", bytes.NewBuffer(jsonBody))
-	if err != nil {
-		t.Fatalf("Failed to create scheduler config request: %v", err)
-	}
-	req.Header.Set("Content-Type", "application/json")
-	resp, err := client.Do(req)
-	if err != nil {
-		t.Fatalf("Failed to update scheduler config: %v", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		body, _ := io.ReadAll(resp.Body)
-		t.Fatalf("Failed to update scheduler config (status %d): %s", resp.StatusCode, string(body))
-	}
-	resp.Body.Close()
-
 	// Disable volume balance to reduce interference for this EC-focused test.
 	balanceConfig := map[string]interface{}{
 		"job_type": "volume_balance",
@@ -223,16 +201,16 @@ func TestEcEndToEnd(t *testing.T) {
 			"enabled": false,
 		},
 	}
-	jsonBody, err = json.Marshal(balanceConfig)
+	jsonBody, err := json.Marshal(balanceConfig)
 	if err != nil {
 		t.Fatalf("Failed to marshal volume_balance config: %v", err)
 	}
-	req, err = http.NewRequest("PUT", AdminUrl+"/api/plugin/job-types/volume_balance/config", bytes.NewBuffer(jsonBody))
+	req, err := http.NewRequest("PUT", AdminUrl+"/api/plugin/job-types/volume_balance/config", bytes.NewBuffer(jsonBody))
 	if err != nil {
 		t.Fatalf("Failed to create volume_balance config request: %v", err)
 	}
 	req.Header.Set("Content-Type", "application/json")
-	resp, err = client.Do(req)
+	resp, err := client.Do(req)
 	if err != nil {
 		t.Fatalf("Failed to update volume_balance config: %v", err)
 	}
@@ -1,6 +1,6 @@
 # Simplified single-stage build for SeaweedFS with FoundationDB support
 # Force x86_64 platform to use AMD64 FoundationDB packages
-FROM --platform=linux/amd64 golang:1.24-bookworm
+FROM --platform=linux/amd64 golang:1.25-bookworm

 ARG FOUNDATIONDB_VERSION=7.4.5
 ENV FOUNDATIONDB_VERSION=${FOUNDATIONDB_VERSION}
@@ -1,5 +1,5 @@
 # Multi-stage Dockerfile to build SeaweedFS with FoundationDB support for ARM64
-FROM --platform=linux/arm64 golang:1.24-bookworm AS builder
+FROM --platform=linux/arm64 golang:1.25-bookworm AS builder

 ARG FOUNDATIONDB_VERSION=7.4.5
 ENV FOUNDATIONDB_VERSION=${FOUNDATIONDB_VERSION}
@@ -1,5 +1,5 @@
 # Test environment with Go and FoundationDB support
-FROM golang:1.24-bookworm
+FROM golang:1.25-bookworm

 # Install system dependencies
 RUN apt-get update && apt-get install -y \
@@ -123,7 +123,7 @@
            <dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
-                <version>3.9.4</version>
+                <version>3.9.5</version>
            </dependency>

            <!-- Apache Commons - Fix CVEs -->
@@ -1,5 +1,5 @@
 # Dockerfile for Kafka Gateway Integration Testing
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Install build dependencies
 RUN apk add --no-cache git make gcc musl-dev sqlite-dev
@@ -1,5 +1,5 @@
 # Dockerfile for building SeaweedFS components from the current workspace
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 RUN apk add --no-cache git make gcc musl-dev sqlite-dev

@@ -1,5 +1,5 @@
 # Dockerfile for Kafka Integration Test Setup
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Install build dependencies
 RUN apk add --no-cache git make gcc musl-dev
@@ -2,7 +2,7 @@
 # Multi-stage build for cross-platform support

 # Stage 1: Builder
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 WORKDIR /app

@@ -27,19 +27,21 @@ type VolumeServer struct {
 	address  string
 	baseDir  string

-	mu                 sync.Mutex
-	receivedFiles      map[string]uint64
-	mountRequests      []*volume_server_pb.VolumeEcShardsMountRequest
-	deleteRequests     []*volume_server_pb.VolumeDeleteRequest
-	markReadonlyCalls  int
-	vacuumGarbageRatio float64
-	vacuumCheckCalls   int
-	vacuumCompactCalls int
-	vacuumCommitCalls  int
-	vacuumCleanupCalls int
-	volumeCopyCalls    int
-	volumeMountCalls   int
-	tailReceiverCalls  int
+	mu                  sync.Mutex
+	receivedFiles       map[string]uint64
+	mountRequests       []*volume_server_pb.VolumeEcShardsMountRequest
+	deleteRequests      []*volume_server_pb.VolumeDeleteRequest
+	markReadonlyCalls   int
+	markWritableCalls   int
+	readFileStatusCalls int
+	vacuumGarbageRatio  float64
+	vacuumCheckCalls    int
+	vacuumCompactCalls  int
+	vacuumCommitCalls   int
+	vacuumCleanupCalls  int
+	volumeCopyCalls     int
+	volumeMountCalls    int
+	tailReceiverCalls   int
 }

 // NewVolumeServer starts a test volume server using the provided base directory.
@@ -151,6 +153,20 @@ func (v *VolumeServer) MarkReadonlyCount() int {
 	return v.markReadonlyCalls
 }

+// MarkWritableCount returns the number of writable calls.
+func (v *VolumeServer) MarkWritableCount() int {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	return v.markWritableCalls
+}
+
+// ReadFileStatusCount returns the number of ReadVolumeFileStatus calls.
+func (v *VolumeServer) ReadFileStatusCount() int {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	return v.readFileStatusCalls
+}
+
 // Shutdown stops the volume server.
 func (v *VolumeServer) Shutdown() {
 	if v.server != nil {
@@ -280,6 +296,25 @@ func (v *VolumeServer) VolumeMarkReadonly(ctx context.Context, req *volume_serve
 	return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil
 }

+func (v *VolumeServer) VolumeMarkWritable(ctx context.Context, req *volume_server_pb.VolumeMarkWritableRequest) (*volume_server_pb.VolumeMarkWritableResponse, error) {
+	v.mu.Lock()
+	v.markWritableCalls++
+	v.mu.Unlock()
+	return &volume_server_pb.VolumeMarkWritableResponse{}, nil
+}
+
+func (v *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) {
+	v.mu.Lock()
+	v.readFileStatusCalls++
+	v.mu.Unlock()
+	return &volume_server_pb.ReadVolumeFileStatusResponse{
+		VolumeId:    req.VolumeId,
+		DatFileSize: 1024,
+		IdxFileSize: 16,
+		FileCount:   1,
+	}, nil
+}
+
 func (v *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_server_pb.VacuumVolumeCheckRequest) (*volume_server_pb.VacuumVolumeCheckResponse, error) {
 	v.mu.Lock()
 	v.vacuumCheckCalls++
@@ -37,7 +37,9 @@ func TestVolumeBalanceDetectionIntegration(t *testing.T) {
 		MasterGrpcAddresses: []string{master.Address()},
 	}, 10)
 	require.NoError(t, err)
-	require.Len(t, proposals, 1)
+	// With default batch_size=20 and 10 overloaded volumes vs 1 underloaded,
+	// all moves are grouped into a single batch proposal.
+	require.Len(t, proposals, 1, "expected exactly one batch proposal")

 	proposal := proposals[0]
 	require.Equal(t, "volume_balance", proposal.JobType)
@@ -46,8 +48,15 @@ func TestVolumeBalanceDetectionIntegration(t *testing.T) {

 	params := &worker_pb.TaskParams{}
 	require.NoError(t, proto.Unmarshal(paramsValue.GetBytesValue(), params))
-	require.NotEmpty(t, params.Sources)
-	require.NotEmpty(t, params.Targets)
+
+	bp := params.GetBalanceParams()
+	require.NotNil(t, bp, "expected BalanceParams in batch proposal")
+	require.Greater(t, len(bp.Moves), 1, "batch proposal should contain multiple moves")
+	for _, move := range bp.Moves {
+		require.NotZero(t, move.VolumeId)
+		require.NotEmpty(t, move.SourceNode)
+		require.NotEmpty(t, move.TargetNode)
+	}
 }

 func buildBalanceVolumeListResponse(t *testing.T) *master_pb.VolumeListResponse {
@@ -8,10 +8,12 @@ import (

 	pluginworkers "github.com/seaweedfs/seaweedfs/test/plugin_workers"
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/proto"
 )

 func TestVolumeBalanceExecutionIntegration(t *testing.T) {
@@ -60,8 +62,92 @@ func TestVolumeBalanceExecutionIntegration(t *testing.T) {
 	require.GreaterOrEqual(t, source.MarkReadonlyCount(), 1)
 	require.GreaterOrEqual(t, len(source.DeleteRequests()), 1)

-	copyCalls, mountCalls, tailCalls := target.BalanceStats()
+	copyCalls, _, tailCalls := target.BalanceStats()
 	require.GreaterOrEqual(t, copyCalls, 1)
-	require.GreaterOrEqual(t, mountCalls, 1)
 	require.GreaterOrEqual(t, tailCalls, 1)
 }
+
+func TestVolumeBalanceBatchExecutionIntegration(t *testing.T) {
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	handler := pluginworker.NewVolumeBalanceHandler(dialOption)
+	harness := pluginworkers.NewHarness(t, pluginworkers.HarnessConfig{
+		WorkerOptions: pluginworker.WorkerOptions{
+			GrpcDialOption: dialOption,
+		},
+		Handlers: []pluginworker.JobHandler{handler},
+	})
+	harness.WaitForJobType("volume_balance")
+
+	// Create one source and one target fake volume server.
+	source := pluginworkers.NewVolumeServer(t, "")
+	target := pluginworkers.NewVolumeServer(t, "")
+
+	// Build a batch job with 3 volume moves from source → target.
+	volumeIDs := []uint32{401, 402, 403}
+	moves := make([]*worker_pb.BalanceMoveSpec, len(volumeIDs))
+	for i, vid := range volumeIDs {
+		moves[i] = &worker_pb.BalanceMoveSpec{
+			VolumeId:   vid,
+			SourceNode: source.Address(),
+			TargetNode: target.Address(),
+			Collection: "batch-test",
+		}
+	}
+
+	params := &worker_pb.TaskParams{
+		TaskId: "batch-balance-test",
+		TaskParams: &worker_pb.TaskParams_BalanceParams{
+			BalanceParams: &worker_pb.BalanceTaskParams{
+				MaxConcurrentMoves: 2,
+				Moves:              moves,
+			},
+		},
+	}
+	paramBytes, err := proto.Marshal(params)
+	require.NoError(t, err)
+
+	job := &plugin_pb.JobSpec{
+		JobId:   "batch-balance-test",
+		JobType: "volume_balance",
+		Parameters: map[string]*plugin_pb.ConfigValue{
+			"task_params_pb": {
+				Kind: &plugin_pb.ConfigValue_BytesValue{BytesValue: paramBytes},
+			},
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	result, err := harness.Plugin().ExecuteJob(ctx, job, nil, 1)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.True(t, result.Success, "batch balance job should succeed; result: %+v", result)
+
+	// Each of the 3 moves should have marked the source readonly and deleted.
+	require.Equal(t, len(volumeIDs), source.MarkReadonlyCount(),
+		"each move should mark source volume readonly")
+	require.Equal(t, len(volumeIDs), len(source.DeleteRequests()),
+		"each move should delete the source volume")
+
+	// Verify delete requests reference the expected volume IDs.
+	deletedVols := make(map[uint32]bool)
+	for _, req := range source.DeleteRequests() {
+		deletedVols[req.VolumeId] = true
+	}
+	for _, vid := range volumeIDs {
+		require.True(t, deletedVols[vid], "volume %d should have been deleted from source", vid)
+	}
+
+	// Pre-delete verification should have called ReadVolumeFileStatus on both
+	// source and target for each volume.
+	require.Equal(t, len(volumeIDs), source.ReadFileStatusCount(),
+		"each move should read source volume status before delete")
+	require.Equal(t, len(volumeIDs), target.ReadFileStatusCount(),
+		"each move should read target volume status before delete")
+
+	// Target should have received copy and tail calls for all 3 volumes.
+	copyCalls, _, tailCalls := target.BalanceStats()
+	require.Equal(t, len(volumeIDs), copyCalls, "target should receive one copy per volume")
+	require.Equal(t, len(volumeIDs), tailCalls, "target should receive one tail per volume")
+}
@@ -1,4 +1,4 @@
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Set working directory
 WORKDIR /app
@@ -1,4 +1,4 @@
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Set working directory
 WORKDIR /app
@@ -1,4 +1,4 @@
-FROM golang:1.24-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Install git and other build dependencies
 RUN apk add --no-cache git make
@@ -140,7 +140,7 @@ func TestCORSConfigurationManagement(t *testing.T) {
 		Bucket:            aws.String(bucketName),
 		CORSConfiguration: corsConfig,
 	})
-	assert.NoError(t, err, "Should be able to put CORS configuration")
+	require.NoError(t, err, "Should be able to put CORS configuration")

 	// Wait for metadata subscription to update cache
 	time.Sleep(50 * time.Millisecond)
@@ -149,9 +149,9 @@ func TestCORSConfigurationManagement(t *testing.T) {
 	getResp, err := client.GetBucketCors(context.TODO(), &s3.GetBucketCorsInput{
 		Bucket: aws.String(bucketName),
 	})
-	assert.NoError(t, err, "Should be able to get CORS configuration")
-	assert.NotNil(t, getResp.CORSRules, "CORS configuration should not be nil")
-	assert.Len(t, getResp.CORSRules, 1, "Should have one CORS rule")
+	require.NoError(t, err, "Should be able to get CORS configuration")
+	require.NotNil(t, getResp.CORSRules, "CORS configuration should not be nil")
+	require.Len(t, getResp.CORSRules, 1, "Should have one CORS rule")

 	rule := getResp.CORSRules[0]
 	assert.Equal(t, []string{"*"}, rule.AllowedHeaders, "Allowed headers should match")
@@ -1,5 +1,5 @@
 # Multi-stage build for SeaweedFS S3 with IAM
-FROM golang:1.23-alpine AS builder
+FROM golang:1.25-alpine AS builder

 # Install build dependencies
 RUN apk add --no-cache git make curl wget
@@ -185,6 +185,9 @@ test-context: ## Test only contextual policy enforcement
 test-presigned: ## Test only presigned URL integration
 	go test -v -run TestS3IAMPresignedURLIntegration ./...

+test-group: ## Run IAM group management tests
+	go test -v -run "TestIAMGroup" ./...
+
 test-sts: ## Run all STS tests
 	go test -v -run "TestSTS" ./...

@@ -263,7 +266,7 @@ docker-build: ## Build custom SeaweedFS image for Docker tests

 # All PHONY targets
 .PHONY: test test-quick run-tests setup start-services stop-services wait-for-services clean logs status debug
-.PHONY: test-auth test-policy test-expiration test-multipart test-bucket-policy test-context test-presigned test-sts test-sts-assume-role test-sts-ldap  
+.PHONY: test-auth test-policy test-expiration test-multipart test-bucket-policy test-context test-presigned test-group test-sts test-sts-assume-role test-sts-ldap  
 .PHONY: benchmark ci watch install-deps docker-test docker-up docker-down docker-logs docker-build
 .PHONY: test-distributed test-performance test-stress test-versioning-stress test-keycloak-full test-all-previously-skipped setup-all-tests help-advanced

@@ -0,0 +1,792 @@
+package iam
+
+import (
+	"encoding/xml"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestIAMGroupLifecycle tests the full lifecycle of group management:
+// CreateGroup, GetGroup, ListGroups, DeleteGroup
+func TestIAMGroupLifecycle(t *testing.T) {
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	groupName := "test-group-lifecycle"
+
+	t.Run("create_group", func(t *testing.T) {
+		resp, err := iamClient.CreateGroup(&iam.CreateGroupInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, groupName, *resp.Group.GroupName)
+	})
+
+	t.Run("get_group", func(t *testing.T) {
+		resp, err := iamClient.GetGroup(&iam.GetGroupInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, groupName, *resp.Group.GroupName)
+	})
+
+	t.Run("list_groups_contains_created", func(t *testing.T) {
+		resp, err := iamClient.ListGroups(&iam.ListGroupsInput{})
+		require.NoError(t, err)
+		found := false
+		for _, g := range resp.Groups {
+			if *g.GroupName == groupName {
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Created group should appear in ListGroups")
+	})
+
+	t.Run("create_duplicate_group_fails", func(t *testing.T) {
+		_, err := iamClient.CreateGroup(&iam.CreateGroupInput{
+			GroupName: aws.String(groupName),
+		})
+		assert.Error(t, err, "Creating a duplicate group should fail")
+	})
+
+	t.Run("delete_group", func(t *testing.T) {
+		_, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+
+		// Verify it's gone
+		resp, err := iamClient.ListGroups(&iam.ListGroupsInput{})
+		require.NoError(t, err)
+		for _, g := range resp.Groups {
+			assert.NotEqual(t, groupName, *g.GroupName,
+				"Deleted group should not appear in ListGroups")
+		}
+	})
+
+	t.Run("delete_nonexistent_group_fails", func(t *testing.T) {
+		_, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{
+			GroupName: aws.String("nonexistent-group-xyz"),
+		})
+		assert.Error(t, err)
+	})
+}
+
+// TestIAMGroupMembership tests adding and removing users from groups
+func TestIAMGroupMembership(t *testing.T) {
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	groupName := "test-group-members"
+	userName := "test-user-for-group"
+
+	// Setup: create group and user
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{
+		GroupName: aws.String(groupName),
+	})
+	require.NoError(t, err)
+	defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)})
+
+	_, err = iamClient.CreateUser(&iam.CreateUserInput{
+		UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+	defer iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
+
+	t.Run("add_user_to_group", func(t *testing.T) {
+		_, err := iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(userName),
+		})
+		require.NoError(t, err)
+	})
+
+	t.Run("get_group_shows_member", func(t *testing.T) {
+		resp, err := iamClient.GetGroup(&iam.GetGroupInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+		found := false
+		for _, u := range resp.Users {
+			if *u.UserName == userName {
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Added user should appear in GetGroup members")
+	})
+
+	t.Run("list_groups_for_user", func(t *testing.T) {
+		resp, err := iamClient.ListGroupsForUser(&iam.ListGroupsForUserInput{
+			UserName: aws.String(userName),
+		})
+		require.NoError(t, err)
+		found := false
+		for _, g := range resp.Groups {
+			if *g.GroupName == groupName {
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Group should appear in ListGroupsForUser")
+	})
+
+	t.Run("add_duplicate_member_is_idempotent", func(t *testing.T) {
+		_, err := iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(userName),
+		})
+		// Should succeed (idempotent) or return a benign error
+		// AWS IAM allows duplicate add without error
+		assert.NoError(t, err)
+	})
+
+	t.Run("remove_user_from_group", func(t *testing.T) {
+		_, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(userName),
+		})
+		require.NoError(t, err)
+
+		// Verify removal
+		resp, err := iamClient.GetGroup(&iam.GetGroupInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+		for _, u := range resp.Users {
+			assert.NotEqual(t, userName, *u.UserName,
+				"Removed user should not appear in group members")
+		}
+	})
+}
+
+// TestIAMGroupPolicyAttachment tests attaching and detaching policies from groups
+func TestIAMGroupPolicyAttachment(t *testing.T) {
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	groupName := "test-group-policies"
+	policyName := "test-group-attach-policy"
+	policyDoc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}]}`
+
+	// Setup: create group and policy
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{
+		GroupName: aws.String(groupName),
+	})
+	require.NoError(t, err)
+
+	createPolicyResp, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
+		PolicyName:     aws.String(policyName),
+		PolicyDocument: aws.String(policyDoc),
+	})
+	require.NoError(t, err)
+	policyArn := createPolicyResp.Policy.Arn
+
+	// Cleanup in correct order: detach policy, delete group, delete policy
+	t.Cleanup(func() {
+		if _, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: policyArn,
+		}); err != nil {
+			t.Logf("cleanup: failed to detach group policy: %v", err)
+		}
+		if _, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)}); err != nil {
+			t.Logf("cleanup: failed to delete group: %v", err)
+		}
+		if _, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: policyArn}); err != nil {
+			t.Logf("cleanup: failed to delete policy: %v", err)
+		}
+	})
+
+	t.Run("attach_group_policy", func(t *testing.T) {
+		_, err := iamClient.AttachGroupPolicy(&iam.AttachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: policyArn,
+		})
+		require.NoError(t, err)
+	})
+
+	t.Run("list_attached_group_policies", func(t *testing.T) {
+		resp, err := iamClient.ListAttachedGroupPolicies(&iam.ListAttachedGroupPoliciesInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+		found := false
+		for _, p := range resp.AttachedPolicies {
+			if *p.PolicyName == policyName {
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Attached policy should appear in ListAttachedGroupPolicies")
+	})
+
+	t.Run("detach_group_policy", func(t *testing.T) {
+		_, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: policyArn,
+		})
+		require.NoError(t, err)
+
+		// Verify detachment
+		resp, err := iamClient.ListAttachedGroupPolicies(&iam.ListAttachedGroupPoliciesInput{
+			GroupName: aws.String(groupName),
+		})
+		require.NoError(t, err)
+		for _, p := range resp.AttachedPolicies {
+			assert.NotEqual(t, policyName, *p.PolicyName,
+				"Detached policy should not appear in ListAttachedGroupPolicies")
+		}
+	})
+}
+
+// TestIAMGroupPolicyEnforcement tests that group policies are enforced during S3 operations.
+// Creates a user with no direct policies, adds them to a group with S3 access,
+// and verifies they can access S3 through the group policy.
+func TestIAMGroupPolicyEnforcement(t *testing.T) {
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	groupName := "test-enforcement-group"
+	userName := "test-enforcement-user"
+	policyName := "test-enforcement-policy"
+	bucketName := "test-group-enforce-bucket"
+	policyDoc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::` + bucketName + `","arn:aws:s3:::` + bucketName + `/*"]}]}`
+
+	// Create user
+	_, err = iamClient.CreateUser(&iam.CreateUserInput{
+		UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+
+	// Create access key for the user
+	keyResp, err := iamClient.CreateAccessKey(&iam.CreateAccessKeyInput{
+		UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+
+	accessKeyId := *keyResp.AccessKey.AccessKeyId
+	secretKey := *keyResp.AccessKey.SecretAccessKey
+
+	// Create an S3 client with the user's credentials
+	userS3Client := createS3Client(t, accessKeyId, secretKey)
+
+	// Create group
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{
+		GroupName: aws.String(groupName),
+	})
+	require.NoError(t, err)
+
+	// Create policy
+	createPolicyResp, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
+		PolicyName:     aws.String(policyName),
+		PolicyDocument: aws.String(policyDoc),
+	})
+	require.NoError(t, err)
+	policyArn := createPolicyResp.Policy.Arn
+
+	// Cleanup in correct order: remove user from group, detach policy,
+	// delete access key, delete user, delete group, delete policy
+	t.Cleanup(func() {
+		if _, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(userName),
+		}); err != nil {
+			t.Logf("cleanup: failed to remove user from group: %v", err)
+		}
+		if _, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: policyArn,
+		}); err != nil {
+			t.Logf("cleanup: failed to detach group policy: %v", err)
+		}
+		if _, err := iamClient.DeleteAccessKey(&iam.DeleteAccessKeyInput{
+			UserName:    aws.String(userName),
+			AccessKeyId: keyResp.AccessKey.AccessKeyId,
+		}); err != nil {
+			t.Logf("cleanup: failed to delete access key: %v", err)
+		}
+		if _, err := iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)}); err != nil {
+			t.Logf("cleanup: failed to delete user: %v", err)
+		}
+		if _, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)}); err != nil {
+			t.Logf("cleanup: failed to delete group: %v", err)
+		}
+		if _, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: policyArn}); err != nil {
+			t.Logf("cleanup: failed to delete policy: %v", err)
+		}
+	})
+
+	// Register bucket cleanup on parent test with admin credentials
+	// (userS3Client may lack permissions by cleanup time)
+	adminS3, err := framework.CreateS3ClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if _, err := adminS3.DeleteObject(&s3.DeleteObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String("test-key"),
+		}); err != nil {
+			t.Logf("cleanup: failed to delete object: %v", err)
+		}
+		if _, err := adminS3.DeleteBucket(&s3.DeleteBucketInput{Bucket: aws.String(bucketName)}); err != nil {
+			t.Logf("cleanup: failed to delete bucket: %v", err)
+		}
+	})
+
+	t.Run("user_without_group_denied", func(t *testing.T) {
+		// User has no policies and is not in any group — should be denied
+		_, err := userS3Client.CreateBucket(&s3.CreateBucketInput{
+			Bucket: aws.String(bucketName),
+		})
+		require.Error(t, err, "User without any policies should be denied")
+		awsErr, ok := err.(awserr.Error)
+		require.True(t, ok, "Expected awserr.Error")
+		assert.Equal(t, "AccessDenied", awsErr.Code())
+	})
+
+	t.Run("user_with_group_policy_allowed", func(t *testing.T) {
+		// Attach policy to group
+		_, err := iamClient.AttachGroupPolicy(&iam.AttachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: policyArn,
+		})
+		require.NoError(t, err)
+
+		// Add user to group
+		_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(userName),
+		})
+		require.NoError(t, err)
+
+		// Wait for policy propagation, then create bucket
+		require.Eventually(t, func() bool {
+			_, err = userS3Client.CreateBucket(&s3.CreateBucketInput{
+				Bucket: aws.String(bucketName),
+			})
+			return err == nil
+		}, 10*time.Second, 500*time.Millisecond, "User with group policy should be allowed")
+
+		// Should also be able to put/get objects
+		_, err = userS3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String("test-key"),
+			Body:   aws.ReadSeekCloser(strings.NewReader("test-data")),
+		})
+		require.NoError(t, err, "User should be able to put objects through group policy")
+	})
+
+	t.Run("user_removed_from_group_denied", func(t *testing.T) {
+		// Remove user from group
+		_, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(userName),
+		})
+		require.NoError(t, err)
+
+		// Wait for policy propagation — user should now be denied
+		var lastErr error
+		require.Eventually(t, func() bool {
+			_, lastErr = userS3Client.ListObjects(&s3.ListObjectsInput{
+				Bucket: aws.String(bucketName),
+			})
+			return lastErr != nil
+		}, 10*time.Second, 500*time.Millisecond, "User removed from group should be denied")
+		awsErr, ok := lastErr.(awserr.Error)
+		require.True(t, ok, "Expected awserr.Error")
+		assert.Equal(t, "AccessDenied", awsErr.Code())
+	})
+}
+
+// TestIAMGroupDisabledPolicyEnforcement tests that disabled groups do not contribute policies.
+// Uses the raw IAM API (callIAMAPI) since the AWS SDK doesn't support custom group status.
+func TestIAMGroupDisabledPolicyEnforcement(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	if !isSeaweedFSRunning(t) {
+		t.Skip("SeaweedFS is not running at", TestIAMEndpoint)
+	}
+
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	groupName := "test-disabled-group"
+	userName := "test-disabled-grp-user"
+	policyName := "test-disabled-grp-policy"
+	bucketName := "test-disabled-grp-bucket"
+	policyDoc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::` + bucketName + `","arn:aws:s3:::` + bucketName + `/*"]}]}`
+
+	// Create user, group, policy
+	_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
+	require.NoError(t, err)
+
+	keyResp, err := iamClient.CreateAccessKey(&iam.CreateAccessKeyInput{UserName: aws.String(userName)})
+	require.NoError(t, err)
+
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(groupName)})
+	require.NoError(t, err)
+
+	createPolicyResp, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
+		PolicyName: aws.String(policyName), PolicyDocument: aws.String(policyDoc),
+	})
+	require.NoError(t, err)
+
+	// Cleanup in correct order: remove user from group, detach policy,
+	// delete access key, delete user, delete group, delete policy
+	t.Cleanup(func() {
+		if _, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(groupName), UserName: aws.String(userName),
+		}); err != nil {
+			t.Logf("cleanup: failed to remove user from group: %v", err)
+		}
+		if _, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: aws.String("arn:aws:iam:::policy/" + policyName),
+		}); err != nil {
+			t.Logf("cleanup: failed to detach group policy: %v", err)
+		}
+		if _, err := iamClient.DeleteAccessKey(&iam.DeleteAccessKeyInput{
+			UserName: aws.String(userName), AccessKeyId: keyResp.AccessKey.AccessKeyId,
+		}); err != nil {
+			t.Logf("cleanup: failed to delete access key: %v", err)
+		}
+		if _, err := iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)}); err != nil {
+			t.Logf("cleanup: failed to delete user: %v", err)
+		}
+		if _, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)}); err != nil {
+			t.Logf("cleanup: failed to delete group: %v", err)
+		}
+		if _, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: createPolicyResp.Policy.Arn}); err != nil {
+			t.Logf("cleanup: failed to delete policy: %v", err)
+		}
+	})
+
+	// Setup: attach policy, add user, create bucket with admin
+	_, err = iamClient.AttachGroupPolicy(&iam.AttachGroupPolicyInput{
+		GroupName: aws.String(groupName), PolicyArn: createPolicyResp.Policy.Arn,
+	})
+	require.NoError(t, err)
+
+	_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+		GroupName: aws.String(groupName), UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+
+	userS3Client := createS3Client(t, *keyResp.AccessKey.AccessKeyId, *keyResp.AccessKey.SecretAccessKey)
+
+	// Create bucket using admin first so we can test listing
+	adminS3, err := framework.CreateS3ClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+	_, err = adminS3.CreateBucket(&s3.CreateBucketInput{Bucket: aws.String(bucketName)})
+	require.NoError(t, err)
+	defer adminS3.DeleteBucket(&s3.DeleteBucketInput{Bucket: aws.String(bucketName)})
+
+	t.Run("enabled_group_allows_access", func(t *testing.T) {
+		require.Eventually(t, func() bool {
+			_, err := userS3Client.ListObjects(&s3.ListObjectsInput{
+				Bucket: aws.String(bucketName),
+			})
+			return err == nil
+		}, 10*time.Second, 500*time.Millisecond, "User in enabled group should have access")
+	})
+
+	t.Run("disabled_group_denies_access", func(t *testing.T) {
+		// Disable group via raw IAM API (no SDK support for this extension)
+		resp, err := callIAMAPIAuthenticated(t, framework, "UpdateGroup", url.Values{
+			"GroupName": {groupName},
+			"Disabled":  {"true"},
+		})
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		require.Equal(t, http.StatusOK, resp.StatusCode, "UpdateGroup (disable) should return 200")
+
+		// Wait for propagation — user should be denied
+		var lastErr error
+		require.Eventually(t, func() bool {
+			_, lastErr = userS3Client.ListObjects(&s3.ListObjectsInput{
+				Bucket: aws.String(bucketName),
+			})
+			return lastErr != nil
+		}, 10*time.Second, 500*time.Millisecond, "User in disabled group should be denied access")
+		awsErr, ok := lastErr.(awserr.Error)
+		require.True(t, ok, "Expected awserr.Error")
+		assert.Equal(t, "AccessDenied", awsErr.Code())
+	})
+
+	t.Run("re_enabled_group_restores_access", func(t *testing.T) {
+		// Re-enable the group
+		resp, err := callIAMAPIAuthenticated(t, framework, "UpdateGroup", url.Values{
+			"GroupName": {groupName},
+			"Disabled":  {"false"},
+		})
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		require.Equal(t, http.StatusOK, resp.StatusCode, "UpdateGroup (re-enable) should return 200")
+
+		// Wait for propagation — user should have access again
+		require.Eventually(t, func() bool {
+			_, err = userS3Client.ListObjects(&s3.ListObjectsInput{
+				Bucket: aws.String(bucketName),
+			})
+			return err == nil
+		}, 10*time.Second, 500*time.Millisecond, "User in re-enabled group should have access again")
+	})
+}
+
+// TestIAMGroupUserDeletionSideEffect tests that deleting a user removes them from all groups.
+func TestIAMGroupUserDeletionSideEffect(t *testing.T) {
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	groupName := "test-deletion-group"
+	userName := "test-deletion-user"
+
+	// Create group and user
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(groupName)})
+	require.NoError(t, err)
+	defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)})
+
+	_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		// Best-effort: user may already be deleted by the test
+		iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
+	})
+
+	// Add user to group
+	_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+		GroupName: aws.String(groupName),
+		UserName:  aws.String(userName),
+	})
+	require.NoError(t, err)
+
+	// Verify user is in group
+	getResp, err := iamClient.GetGroup(&iam.GetGroupInput{GroupName: aws.String(groupName)})
+	require.NoError(t, err)
+	assert.Len(t, getResp.Users, 1, "Group should have 1 member before deletion")
+
+	// Delete the user
+	_, err = iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
+	require.NoError(t, err)
+
+	// Verify user was removed from the group
+	getResp, err = iamClient.GetGroup(&iam.GetGroupInput{GroupName: aws.String(groupName)})
+	require.NoError(t, err)
+	assert.Empty(t, getResp.Users, "Group should have no members after user deletion")
+}
+
+// TestIAMGroupMultipleGroups tests that a user can belong to multiple groups
+// and inherits policies from all of them.
+func TestIAMGroupMultipleGroups(t *testing.T) {
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
+	require.NoError(t, err)
+
+	group1 := "test-multi-group-1"
+	group2 := "test-multi-group-2"
+	userName := "test-multi-group-user"
+
+	// Create two groups
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(group1)})
+	require.NoError(t, err)
+	defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(group1)})
+
+	_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(group2)})
+	require.NoError(t, err)
+	defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(group2)})
+
+	// Create user
+	_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
+	require.NoError(t, err)
+	defer func() {
+		iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(group1), UserName: aws.String(userName),
+		})
+		iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(group2), UserName: aws.String(userName),
+		})
+		iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
+	}()
+
+	// Add user to both groups
+	_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+		GroupName: aws.String(group1), UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+
+	_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
+		GroupName: aws.String(group2), UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+
+	// Verify user appears in both groups
+	resp, err := iamClient.ListGroupsForUser(&iam.ListGroupsForUserInput{
+		UserName: aws.String(userName),
+	})
+	require.NoError(t, err)
+	groupNames := make(map[string]bool)
+	for _, g := range resp.Groups {
+		groupNames[*g.GroupName] = true
+	}
+	assert.True(t, groupNames[group1], "User should be in group 1")
+	assert.True(t, groupNames[group2], "User should be in group 2")
+}
+
+// --- Response types for raw IAM API calls ---
+
+type CreateGroupResponse struct {
+	XMLName           xml.Name `xml:"CreateGroupResponse"`
+	CreateGroupResult struct {
+		Group struct {
+			GroupName string `xml:"GroupName"`
+		} `xml:"Group"`
+	} `xml:"CreateGroupResult"`
+}
+
+type ListGroupsResponse struct {
+	XMLName          xml.Name `xml:"ListGroupsResponse"`
+	ListGroupsResult struct {
+		Groups []struct {
+			GroupName string `xml:"GroupName"`
+		} `xml:"Groups>member"`
+	} `xml:"ListGroupsResult"`
+}
+
+// callIAMAPIAuthenticated sends an authenticated raw IAM API request using the
+// framework's JWT token. This is needed for custom extensions not in the AWS SDK
+// (like UpdateGroup with Disabled parameter).
+func callIAMAPIAuthenticated(_ *testing.T, framework *S3IAMTestFramework, action string, params url.Values) (*http.Response, error) {
+	params.Set("Action", action)
+
+	req, err := http.NewRequest(http.MethodPost, TestIAMEndpoint+"/",
+		strings.NewReader(params.Encode()))
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	token, err := framework.generateSTSSessionToken("admin-user", "TestAdminRole", time.Hour, "", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	client := &http.Client{
+		Timeout:   30 * time.Second,
+		Transport: &BearerTokenTransport{Token: token},
+	}
+	return client.Do(req)
+}
+
+// TestIAMGroupRawAPI tests group operations using raw HTTP IAM API calls,
+// verifying XML response format for group operations.
+func TestIAMGroupRawAPI(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	if !isSeaweedFSRunning(t) {
+		t.Skip("SeaweedFS is not running at", TestIAMEndpoint)
+	}
+
+	framework := NewS3IAMTestFramework(t)
+	defer framework.Cleanup()
+
+	groupName := "test-raw-api-group"
+
+	t.Run("create_group_raw", func(t *testing.T) {
+		resp, err := callIAMAPIAuthenticated(t, framework, "CreateGroup", url.Values{
+			"GroupName": {groupName},
+		})
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+		body, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+
+		var createResp CreateGroupResponse
+		err = xml.Unmarshal(body, &createResp)
+		require.NoError(t, err)
+		assert.Equal(t, groupName, createResp.CreateGroupResult.Group.GroupName)
+	})
+
+	t.Run("list_groups_raw", func(t *testing.T) {
+		resp, err := callIAMAPIAuthenticated(t, framework, "ListGroups", url.Values{})
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+		body, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+
+		var listResp ListGroupsResponse
+		err = xml.Unmarshal(body, &listResp)
+		require.NoError(t, err)
+
+		found := false
+		for _, g := range listResp.ListGroupsResult.Groups {
+			if g.GroupName == groupName {
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Created group should appear in raw ListGroups")
+	})
+
+	t.Run("delete_group_raw", func(t *testing.T) {
+		resp, err := callIAMAPIAuthenticated(t, framework, "DeleteGroup", url.Values{
+			"GroupName": {groupName},
+		})
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+	})
+}
+
+// createS3Client creates an S3 client with static credentials
+func createS3Client(t *testing.T, accessKey, secretKey string) *s3.S3 {
+	sess, err := session.NewSession(&aws.Config{
+		Region:           aws.String("us-east-1"),
+		Endpoint:         aws.String(TestS3Endpoint),
+		Credentials:      credentials.NewStaticCredentials(accessKey, secretKey, ""),
+		DisableSSL:       aws.Bool(true),
+		S3ForcePathStyle: aws.Bool(true),
+	})
+	require.NoError(t, err)
+	return s3.New(sess)
+}
@@ -10,14 +10,14 @@ import (
 	"testing"
 	"time"

-	"github.com/aws/aws-sdk-go/aws"
-	v1credentials "github.com/aws/aws-sdk-go/aws/credentials"
-	v1signer "github.com/aws/aws-sdk-go/aws/signer/v4"
-	v1s3 "github.com/aws/aws-sdk-go/service/s3"
 	v2aws "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/credentials"
 	v2s3 "github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/aws/aws-sdk-go/aws"
+	v1credentials "github.com/aws/aws-sdk-go/aws/credentials"
+	v1signer "github.com/aws/aws-sdk-go/aws/signer/v4"
+	v1s3 "github.com/aws/aws-sdk-go/service/s3"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -187,8 +187,8 @@ func testGetObjectAttributesMultipart(t *testing.T, cluster *TestCluster) {

 	// Test pagination: MaxParts=1
 	resp2, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
-		Bucket:  v2aws.String(bucketName),
-		Key:     v2aws.String(objectKey),
+		Bucket:   v2aws.String(bucketName),
+		Key:      v2aws.String(objectKey),
 		MaxParts: v2aws.Int32(1),
 		ObjectAttributes: []types.ObjectAttributes{
 			types.ObjectAttributesObjectParts,
@@ -0,0 +1,987 @@
+// Package maintenance contains integration tests for the iceberg table
+// maintenance plugin worker. Tests start a real weed mini cluster, create
+// tables via the S3 Tables API, populate Iceberg metadata via the filer
+// gRPC API, and then exercise the iceberg.Handler operations against the
+// live filer.
+package maintenance
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/parquet-go/parquet-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/command"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	icebergHandler "github.com/seaweedfs/seaweedfs/weed/plugin/worker/iceberg"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// ---------------------------------------------------------------------------
+// Cluster lifecycle (mirrors test/s3tables/table-buckets pattern)
+// ---------------------------------------------------------------------------
+
+type testCluster struct {
+	dataDir       string
+	ctx           context.Context
+	cancel        context.CancelFunc
+	wg            sync.WaitGroup
+	filerGrpcPort int
+	s3Port        int
+	s3Endpoint    string
+	isRunning     bool
+}
+
+var shared *testCluster
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+	if testing.Short() {
+		os.Exit(m.Run())
+	}
+
+	testDir, err := os.MkdirTemp("", "seaweed-iceberg-maint-*")
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "SKIP: failed to create temp dir: %v\n", err)
+		os.Exit(0)
+	}
+
+	cluster, err := startCluster(testDir, nil)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "SKIP: failed to start cluster: %v\n", err)
+		os.RemoveAll(testDir)
+		os.Exit(0)
+	}
+	shared = cluster
+
+	code := m.Run()
+	shared.stop()
+	os.RemoveAll(testDir)
+	os.Exit(code)
+}
+
+func startCluster(testDir string, extraArgs []string) (*testCluster, error) {
+	ports, err := findPorts(10)
+	if err != nil {
+		return nil, err
+	}
+	masterPort, masterGrpc := ports[0], ports[1]
+	volumePort, volumeGrpc := ports[2], ports[3]
+	filerPort, filerGrpc := ports[4], ports[5]
+	s3Port, s3Grpc := ports[6], ports[7]
+	adminPort, adminGrpc := ports[8], ports[9]
+
+	_ = os.Remove(filepath.Join(testDir, "mini.options"))
+
+	// Empty security.toml disables JWT auth.
+	if err := os.WriteFile(filepath.Join(testDir, "security.toml"), []byte("# test\n"), 0644); err != nil {
+		return nil, err
+	}
+	if os.Getenv("AWS_ACCESS_KEY_ID") == "" {
+		os.Setenv("AWS_ACCESS_KEY_ID", "admin")
+	}
+	if os.Getenv("AWS_SECRET_ACCESS_KEY") == "" {
+		os.Setenv("AWS_SECRET_ACCESS_KEY", "admin")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	c := &testCluster{
+		dataDir:       testDir,
+		ctx:           ctx,
+		cancel:        cancel,
+		filerGrpcPort: filerGrpc,
+		s3Port:        s3Port,
+		s3Endpoint:    fmt.Sprintf("http://127.0.0.1:%d", s3Port),
+	}
+
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		oldDir, _ := os.Getwd()
+		oldArgs := os.Args
+		defer func() { os.Chdir(oldDir); os.Args = oldArgs }()
+		os.Chdir(testDir)
+
+		args := []string{
+			"-dir=" + testDir,
+			"-master.dir=" + testDir,
+			"-master.port=" + strconv.Itoa(masterPort),
+			"-master.port.grpc=" + strconv.Itoa(masterGrpc),
+			"-volume.port=" + strconv.Itoa(volumePort),
+			"-volume.port.grpc=" + strconv.Itoa(volumeGrpc),
+			"-volume.port.public=" + strconv.Itoa(volumePort),
+			"-volume.publicUrl=127.0.0.1:" + strconv.Itoa(volumePort),
+			"-filer.port=" + strconv.Itoa(filerPort),
+			"-filer.port.grpc=" + strconv.Itoa(filerGrpc),
+			"-s3.port=" + strconv.Itoa(s3Port),
+			"-s3.port.grpc=" + strconv.Itoa(s3Grpc),
+			"-admin.port=" + strconv.Itoa(adminPort),
+			"-admin.port.grpc=" + strconv.Itoa(adminGrpc),
+			"-webdav.port=0",
+			"-admin.ui=false",
+			"-master.volumeSizeLimitMB=32",
+			"-ip=127.0.0.1",
+			"-master.peers=none",
+			"-s3.iam.readOnly=false",
+		}
+		args = append(args, extraArgs...)
+		os.Args = append([]string{"weed"}, args...)
+		glog.MaxSize = 1024 * 1024
+		for _, cmd := range command.Commands {
+			if cmd.Name() == "mini" && cmd.Run != nil {
+				cmd.Flag.Parse(os.Args[1:])
+				command.MiniClusterCtx = ctx
+				cmd.Run(cmd, cmd.Flag.Args())
+				command.MiniClusterCtx = nil
+				return
+			}
+		}
+	}()
+
+	if err := waitReady(c.s3Endpoint, 30*time.Second); err != nil {
+		cancel()
+		return nil, err
+	}
+	c.isRunning = true
+	return c, nil
+}
+
+func (c *testCluster) stop() {
+	if c.cancel != nil {
+		c.cancel()
+	}
+	if c.isRunning {
+		time.Sleep(500 * time.Millisecond)
+	}
+	done := make(chan struct{})
+	go func() { c.wg.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+	}
+}
+
+func (c *testCluster) filerConn(t *testing.T) (*grpc.ClientConn, filer_pb.SeaweedFilerClient) {
+	t.Helper()
+	addr := fmt.Sprintf("127.0.0.1:%d", c.filerGrpcPort)
+	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { conn.Close() })
+	return conn, filer_pb.NewSeaweedFilerClient(conn)
+}
+
+func findPorts(n int) ([]int, error) {
+	ls := make([]*net.TCPListener, n)
+	ps := make([]int, n)
+	for i := 0; i < n; i++ {
+		l, err := net.Listen("tcp", "127.0.0.1:0")
+		if err != nil {
+			for j := 0; j < i; j++ {
+				ls[j].Close()
+			}
+			return nil, err
+		}
+		ls[i] = l.(*net.TCPListener)
+		ps[i] = ls[i].Addr().(*net.TCPAddr).Port
+	}
+	for _, l := range ls {
+		l.Close()
+	}
+	return ps, nil
+}
+
+func waitReady(endpoint string, timeout time.Duration) error {
+	client := &http.Client{Timeout: 1 * time.Second}
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		resp, err := client.Get(endpoint)
+		if err == nil {
+			resp.Body.Close()
+			time.Sleep(500 * time.Millisecond)
+			return nil
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return fmt.Errorf("timeout waiting for %s", endpoint)
+}
+
+func randomSuffix() string {
+	b := make([]byte, 4)
+	rand.Read(b)
+	return fmt.Sprintf("%x", b)
+}
+
+// ---------------------------------------------------------------------------
+// Helpers for populating Iceberg table state via filer gRPC
+// ---------------------------------------------------------------------------
+
+func newTestSchema() *iceberg.Schema {
+	return iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true},
+		iceberg.NestedField{ID: 2, Type: iceberg.PrimitiveTypes.String, Name: "name", Required: false},
+		iceberg.NestedField{ID: 3, Type: iceberg.PrimitiveTypes.String, Name: "payload", Required: false},
+	)
+}
+
+type compactRow struct {
+	ID      int64  `parquet:"id"`
+	Name    string `parquet:"name"`
+	Payload string `parquet:"payload"`
+}
+
+func buildParquetData(t *testing.T, prefix string, rowCount int) []byte {
+	t.Helper()
+
+	rows := make([]compactRow, rowCount)
+	for i := range rows {
+		rows[i] = compactRow{
+			ID:      int64(i + 1),
+			Name:    fmt.Sprintf("%s-%06d", prefix, i),
+			Payload: fmt.Sprintf("%s-%06d-%s", prefix, i, strings.Repeat(fmt.Sprintf("%04d", i%10000), 128)),
+		}
+	}
+
+	var buf bytes.Buffer
+	writer := parquet.NewGenericWriter[compactRow](&buf)
+	_, err := writer.Write(rows)
+	require.NoError(t, err)
+	require.NoError(t, writer.Close())
+	return buf.Bytes()
+}
+
+func buildMetadata(t *testing.T, snapshots []table.Snapshot) table.Metadata {
+	t.Helper()
+	schema := newTestSchema()
+	meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, "s3://test/table", nil)
+	require.NoError(t, err)
+	if len(snapshots) == 0 {
+		return meta
+	}
+
+	// Iceberg validates that snapshot timestamps >= metadata lastUpdatedMs.
+	// Offset all timestamps so they're safely after the metadata was created.
+	baseMs := time.Now().UnixMilli() + 100
+	for i := range snapshots {
+		snapshots[i].TimestampMs = baseMs + int64(i)
+	}
+
+	builder, err := table.MetadataBuilderFromBase(meta, "s3://test/table")
+	require.NoError(t, err)
+	var lastID int64
+	for i := range snapshots {
+		s := snapshots[i]
+		require.NoError(t, builder.AddSnapshot(&s))
+		lastID = s.SnapshotID
+	}
+	require.NoError(t, builder.SetSnapshotRef(table.MainBranch, lastID, table.BranchRef))
+	result, err := builder.Build()
+	require.NoError(t, err)
+	return result
+}
+
+// s3put performs an S3-signed PUT request. When key is empty it creates a
+// bucket; otherwise it uploads an object.
+func s3put(t *testing.T, s3Endpoint, bucket, key string, body []byte) {
+	t.Helper()
+	urlPath := "/" + bucket
+	if key != "" {
+		urlPath += "/" + key
+	}
+	var reader *bytes.Reader
+	if body != nil {
+		reader = bytes.NewReader(body)
+	} else {
+		reader = bytes.NewReader(nil)
+	}
+	req, err := http.NewRequest(http.MethodPut, s3Endpoint+urlPath, reader)
+	require.NoError(t, err)
+	req.Header.Set("Host", req.URL.Host)
+	hash := sha256.Sum256(body)
+	err = v4.NewSigner().SignHTTP(context.Background(),
+		aws.Credentials{AccessKeyID: "admin", SecretAccessKey: "admin"},
+		req, hex.EncodeToString(hash[:]), "s3", "us-east-1", time.Now())
+	require.NoError(t, err)
+	resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	if resp.StatusCode >= 400 {
+		b, _ := io.ReadAll(resp.Body)
+		require.FailNowf(t, "s3put failed", "s3put(%s/%s) → %d: %s", bucket, key, resp.StatusCode, string(b))
+	}
+}
+
+func newS3Client(t *testing.T, endpoint string) *s3.Client {
+	t.Helper()
+
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion("us-east-1"),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("admin", "admin", "")),
+	)
+	require.NoError(t, err)
+
+	return s3.NewFromConfig(cfg, func(o *s3.Options) {
+		o.BaseEndpoint = aws.String(endpoint)
+		o.UsePathStyle = true
+	})
+}
+
+func s3putObject(t *testing.T, client *s3.Client, bucket, key string, body []byte) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+
+	_, err := client.PutObject(ctx, &s3.PutObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+		Body:   bytes.NewReader(body),
+	})
+	require.NoError(t, err)
+}
+
+type testFilerClient struct {
+	client filer_pb.SeaweedFilerClient
+}
+
+func (c testFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c.client)
+}
+
+func (c testFilerClient) AdjustedUrl(location *filer_pb.Location) string {
+	if location == nil {
+		return ""
+	}
+	if location.PublicUrl != "" {
+		return location.PublicUrl
+	}
+	return location.Url
+}
+
+func (c testFilerClient) GetDataCenter() string {
+	return ""
+}
+
+func readFile(t *testing.T, client filer_pb.SeaweedFilerClient, dir, name string) []byte {
+	t.Helper()
+
+	entry := lookupEntry(t, client, dir, name)
+	require.NotNil(t, entry, "readFile(%s, %s): entry missing", dir, name)
+	if len(entry.Content) > 0 || len(entry.Chunks) == 0 {
+		return entry.Content
+	}
+
+	reader := filer.NewFileReader(testFilerClient{client: client}, entry)
+	if closer, ok := reader.(io.Closer); ok {
+		defer closer.Close()
+	}
+	content, err := io.ReadAll(reader)
+	require.NoError(t, err, "readFile(%s, %s): read chunked content", dir, name)
+	return content
+}
+
+// createBucketViaS3 creates a regular S3 bucket via the S3 PUT Bucket API.
+// populateTableViaFiler creates the Iceberg directory structure, metadata xattr,
+// manifests, and data files for a table directly in the filer.
+// The table bucket, namespace, and table must already exist (created via S3
+// Tables API).
+func populateTableViaFiler(
+	t *testing.T,
+	client filer_pb.SeaweedFilerClient,
+	s3Endpoint, bucketName, namespace, tableName string,
+	snapshots []table.Snapshot,
+) table.Metadata {
+	t.Helper()
+	ctx := context.Background()
+	meta := buildMetadata(t, snapshots)
+	fullJSON, err := json.Marshal(meta)
+	require.NoError(t, err)
+
+	tablePath := path.Join(namespace, tableName)
+	bucketsPath := s3tables.TablesPath
+	tableFilerPath := path.Join(bucketsPath, bucketName, tablePath)
+	metaDir := path.Join(tableFilerPath, "metadata")
+	dataDir := path.Join(tableFilerPath, "data")
+
+	// Build the table metadata xattr
+	internalMeta := map[string]interface{}{
+		"metadataVersion":  1,
+		"metadataLocation": "metadata/v1.metadata.json",
+		"metadata": map[string]interface{}{
+			"fullMetadata": json.RawMessage(fullJSON),
+		},
+	}
+	xattr, err := json.Marshal(internalMeta)
+	require.NoError(t, err)
+
+	// Create the S3 bucket via PUT, then create the directory tree via S3 PUT
+	// object with zero-byte content (the filer creates intermediate dirs).
+	s3put(t, s3Endpoint, bucketName, "", nil)                              // create bucket
+	s3put(t, s3Endpoint, bucketName, namespace+"/.dir", []byte{})          // create ns dir
+	s3put(t, s3Endpoint, bucketName, tablePath+"/.dir", []byte{})          // create table dir
+	s3put(t, s3Endpoint, bucketName, tablePath+"/metadata/.dir", []byte{}) // create metadata dir
+	s3put(t, s3Endpoint, bucketName, tablePath+"/data/.dir", []byte{})     // create data dir
+
+	// Now set the table bucket and table xattrs via filer gRPC.
+	// Mark bucket as table bucket.
+	bucketEntry := lookupEntry(t, client, bucketsPath, bucketName)
+	require.NotNil(t, bucketEntry, "bucket should exist after S3 PUT")
+	if bucketEntry.Extended == nil {
+		bucketEntry.Extended = make(map[string][]byte)
+	}
+	bucketEntry.Extended[s3tables.ExtendedKeyTableBucket] = []byte("true")
+	_, err = client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
+		Directory: bucketsPath, Entry: bucketEntry,
+	})
+	require.NoError(t, err)
+
+	// Set table metadata xattr.
+	nsDir := path.Join(bucketsPath, bucketName, namespace)
+	tableEntry := lookupEntry(t, client, nsDir, tableName)
+	require.NotNil(t, tableEntry, "table dir should exist after S3 PUT")
+	if tableEntry.Extended == nil {
+		tableEntry.Extended = make(map[string][]byte)
+	}
+	tableEntry.Extended[s3tables.ExtendedKeyMetadata] = xattr
+	_, err = client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
+		Directory: nsDir, Entry: tableEntry,
+	})
+	require.NoError(t, err)
+
+	// Write manifest + manifest list + data file for each snapshot
+	schema := meta.CurrentSchema()
+	spec := meta.PartitionSpec()
+	version := meta.Version()
+
+	for _, snap := range snapshots {
+		if snap.ManifestList == "" {
+			continue
+		}
+		snapID := snap.SnapshotID
+		dataFileName := fmt.Sprintf("snap-%d-data.parquet", snapID)
+		dfBuilder, err := iceberg.NewDataFileBuilder(
+			spec, iceberg.EntryContentData,
+			path.Join("data", dataFileName), iceberg.ParquetFile,
+			map[int]any{}, nil, nil, 10, 4096,
+		)
+		require.NoError(t, err)
+
+		entry := iceberg.NewManifestEntry(
+			iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build(),
+		)
+
+		// Manifest
+		manifestName := fmt.Sprintf("manifest-%d.avro", snapID)
+		var manifestBuf bytes.Buffer
+		mf, err := iceberg.WriteManifest(
+			path.Join("metadata", manifestName), &manifestBuf,
+			version, spec, schema, snapID, []iceberg.ManifestEntry{entry},
+		)
+		require.NoError(t, err)
+		writeFile(t, ctx, client, metaDir, manifestName, manifestBuf.Bytes())
+
+		// Manifest list
+		mlName := path.Base(snap.ManifestList)
+		var mlBuf bytes.Buffer
+		parent := snap.ParentSnapshotID
+		seqNum := snap.SequenceNumber
+		require.NoError(t, iceberg.WriteManifestList(
+			version, &mlBuf, snapID, parent, &seqNum, 0,
+			[]iceberg.ManifestFile{mf},
+		))
+		writeFile(t, ctx, client, metaDir, mlName, mlBuf.Bytes())
+
+		// Data file (dummy content)
+		writeFile(t, ctx, client, dataDir, dataFileName, []byte("fake-parquet"))
+	}
+
+	// Wait for snapshot timestamps (set slightly in the future by
+	// buildMetadata) to move into the past so expiration checks work.
+	time.Sleep(200 * time.Millisecond)
+
+	return meta
+}
+
+func writeFile(t *testing.T, ctx context.Context, client filer_pb.SeaweedFilerClient, dir, name string, content []byte) {
+	t.Helper()
+	resp, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
+		Directory: dir,
+		Entry: &filer_pb.Entry{
+			Name: name,
+			Attributes: &filer_pb.FuseAttributes{
+				Mtime: time.Now().Unix(), Crtime: time.Now().Unix(),
+				FileMode: uint32(0644), FileSize: uint64(len(content)),
+			},
+			Content: content,
+		},
+	})
+	require.NoError(t, err, "writeFile(%s, %s): rpc error", dir, name)
+	require.Empty(t, resp.Error, "writeFile(%s, %s): resp error", dir, name)
+}
+
+func upsertFile(t *testing.T, ctx context.Context, client filer_pb.SeaweedFilerClient, dir, name string, content []byte) {
+	t.Helper()
+
+	entry := lookupEntry(t, client, dir, name)
+	if entry == nil {
+		writeFile(t, ctx, client, dir, name, content)
+		return
+	}
+	if entry.Attributes == nil {
+		entry.Attributes = &filer_pb.FuseAttributes{}
+	}
+	entry.Content = content
+	entry.Chunks = nil
+	entry.Attributes.Mtime = time.Now().Unix()
+	entry.Attributes.FileMode = uint32(0644)
+	entry.Attributes.FileSize = uint64(len(content))
+	resp, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
+		Directory: dir,
+		Entry:     entry,
+	})
+	require.NoError(t, err, "upsertFile(%s, %s): rpc error", dir, name)
+	require.NotNil(t, resp, "upsertFile(%s, %s): nil response", dir, name)
+}
+
+func lookupEntry(t *testing.T, client filer_pb.SeaweedFilerClient, dir, name string) *filer_pb.Entry {
+	t.Helper()
+	resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: dir, Name: name,
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return nil
+		}
+		require.NoError(t, err, "lookupEntry(%s, %s): unexpected error", dir, name)
+	}
+	return resp.Entry
+}
+
+// ---------------------------------------------------------------------------
+// Integration tests
+// ---------------------------------------------------------------------------
+
+func TestIcebergMaintenanceIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	t.Run("ExpireSnapshots", testExpireSnapshots)
+	t.Run("CompactDataFiles", testCompactDataFiles)
+	t.Run("RemoveOrphans", testRemoveOrphans)
+	t.Run("RewriteManifests", testRewriteManifests)
+	t.Run("FullMaintenanceCycle", testFullMaintenanceCycle)
+}
+
+func testExpireSnapshots(t *testing.T) {
+	_, client := shared.filerConn(t)
+	suffix := randomSuffix()
+	bucket := "maint-expire-" + suffix
+	ns := "ns"
+	tbl := "tbl"
+
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+		{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+	}
+	populateTableViaFiler(t, client, shared.s3Endpoint, bucket, ns, tbl, snapshots)
+
+	handler := icebergHandler.NewHandler(nil)
+	config := icebergHandler.Config{
+		SnapshotRetentionHours: 0, // instant expiry — everything eligible
+		MaxSnapshotsToKeep:     1, // keep only the current snapshot
+		MaxCommitRetries:       3,
+	}
+
+	result, _, err := handler.ExpireSnapshots(context.Background(), client, bucket, path.Join(ns, tbl), config)
+	require.NoError(t, err)
+	assert.Contains(t, result, "expired")
+	t.Logf("ExpireSnapshots result: %s", result)
+
+	// Verify metadata was updated
+	entry := lookupEntry(t, client, path.Join(s3tables.TablesPath, bucket, ns), tbl)
+	require.NotNil(t, entry)
+	xattr := entry.Extended[s3tables.ExtendedKeyMetadata]
+	require.NotEmpty(t, xattr)
+
+	// Parse updated metadata to verify snapshot count
+	var internalMeta struct {
+		MetadataVersion int `json:"metadataVersion"`
+	}
+	require.NoError(t, json.Unmarshal(xattr, &internalMeta))
+	assert.Greater(t, internalMeta.MetadataVersion, 1, "metadata version should have been incremented")
+}
+
+func testCompactDataFiles(t *testing.T) {
+	_, client := shared.filerConn(t)
+	suffix := randomSuffix()
+	bucket := "maint-compact-" + suffix
+	ns := "ns"
+	tbl := "tbl"
+
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
+	}
+	meta := populateTableViaFiler(t, client, shared.s3Endpoint, bucket, ns, tbl, snapshots)
+
+	ctx := context.Background()
+	tablePath := path.Join(ns, tbl)
+	dataDir := path.Join(s3tables.TablesPath, bucket, tablePath, "data")
+	metaDir := path.Join(s3tables.TablesPath, bucket, tablePath, "metadata")
+
+	type parquetFile struct {
+		name        string
+		recordCount int64
+		fileSize    int64
+	}
+
+	s3Client := newS3Client(t, shared.s3Endpoint)
+	var files []parquetFile
+	for i := 1; i <= 3; i++ {
+		name := fmt.Sprintf("compact-input-%d.parquet", i)
+		content := buildParquetData(t, fmt.Sprintf("compact-%d", i), 2000)
+		s3putObject(t, s3Client, bucket, path.Join(tablePath, "data", name), content)
+
+		entry := lookupEntry(t, client, dataDir, name)
+		require.NotNil(t, entry, "uploaded parquet file should exist")
+		require.NotEmpty(t, entry.Chunks, "uploaded parquet file should be chunk-backed")
+
+		files = append(files, parquetFile{
+			name:        name,
+			recordCount: 2000,
+			fileSize:    int64(len(content)),
+		})
+	}
+
+	schema := meta.CurrentSchema()
+	spec := meta.PartitionSpec()
+	version := meta.Version()
+	snapID := snapshots[0].SnapshotID
+
+	entries := make([]iceberg.ManifestEntry, 0, len(files))
+	for _, file := range files {
+		dfBuilder, err := iceberg.NewDataFileBuilder(
+			spec,
+			iceberg.EntryContentData,
+			path.Join("data", file.name),
+			iceberg.ParquetFile,
+			map[int]any{},
+			nil, nil,
+			file.recordCount,
+			file.fileSize,
+		)
+		require.NoError(t, err)
+		entries = append(entries, iceberg.NewManifestEntry(
+			iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build(),
+		))
+	}
+
+	manifestName := "manifest-compact.avro"
+	var manifestBuf bytes.Buffer
+	manifest, err := iceberg.WriteManifest(
+		path.Join("metadata", manifestName), &manifestBuf,
+		version, spec, schema, snapID, entries,
+	)
+	require.NoError(t, err)
+	writeFile(t, ctx, client, metaDir, manifestName, manifestBuf.Bytes())
+
+	var manifestListBuf bytes.Buffer
+	seqNum := snapshots[0].SequenceNumber
+	require.NoError(t, iceberg.WriteManifestList(
+		version, &manifestListBuf, snapID, snapshots[0].ParentSnapshotID, &seqNum, 0,
+		[]iceberg.ManifestFile{manifest},
+	))
+	upsertFile(t, ctx, client, metaDir, path.Base(snapshots[0].ManifestList), manifestListBuf.Bytes())
+
+	handler := icebergHandler.NewHandler(nil)
+	config := icebergHandler.Config{
+		TargetFileSizeBytes: 16 * 1024 * 1024,
+		MinInputFiles:       2,
+		MaxCommitRetries:    3,
+	}
+
+	result, metrics, err := handler.CompactDataFiles(ctx, client, bucket, tablePath, config)
+	require.NoError(t, err)
+	assert.Contains(t, result, "compacted")
+	require.NotNil(t, metrics, "expected non-nil metrics from CompactDataFiles")
+	assert.Contains(t, metrics, icebergHandler.MetricFilesMerged)
+	assert.Contains(t, metrics, icebergHandler.MetricFilesWritten)
+	assert.Contains(t, metrics, icebergHandler.MetricBins)
+	assert.Contains(t, metrics, icebergHandler.MetricDurationMs)
+	t.Logf("CompactDataFiles result: %s, metrics: %v", result, metrics)
+
+	var compacted *filer_pb.Entry
+	listErr := filer_pb.SeaweedList(ctx, client, dataDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+		// Compacted output uses compact-{snapID}-{newSnapID}-{binIdx}.parquet.
+		// The test inputs use compact-input-{n}.parquet and must not satisfy this check.
+		if strings.HasPrefix(entry.Name, "compact-") &&
+			!strings.HasPrefix(entry.Name, "compact-input-") &&
+			strings.HasSuffix(entry.Name, ".parquet") {
+			compacted = entry
+		}
+		return nil
+	}, "", false, 1000)
+	require.NoError(t, listErr)
+	require.NotNil(t, compacted, "compaction should create a merged parquet file")
+
+	tableEntry := lookupEntry(t, client, path.Join(s3tables.TablesPath, bucket, ns), tbl)
+	require.NotNil(t, tableEntry)
+	xattr := tableEntry.Extended[s3tables.ExtendedKeyMetadata]
+	require.NotEmpty(t, xattr)
+
+	var internalMeta struct {
+		MetadataVersion  int    `json:"metadataVersion"`
+		MetadataLocation string `json:"metadataLocation,omitempty"`
+	}
+	require.NoError(t, json.Unmarshal(xattr, &internalMeta))
+	assert.Greater(t, internalMeta.MetadataVersion, 1, "metadata version should advance after compaction")
+
+	metadataName := path.Base(internalMeta.MetadataLocation)
+	require.NotEmpty(t, metadataName, "compaction should point table metadata at a committed metadata file")
+	metadataBytes := readFile(t, client, metaDir, metadataName)
+	committedMeta, err := table.ParseMetadataBytes(metadataBytes)
+	require.NoError(t, err)
+
+	currentSnap := committedMeta.CurrentSnapshot()
+	require.NotNil(t, currentSnap, "committed metadata should include a current snapshot")
+	assert.Equal(t, snapshots[0].SequenceNumber+1, currentSnap.SequenceNumber, "compaction should advance the snapshot sequence number")
+
+	manifestListName := path.Base(currentSnap.ManifestList)
+	require.NotEmpty(t, manifestListName, "committed snapshot should reference a manifest list")
+	manifestListBytes := readFile(t, client, metaDir, manifestListName)
+	manifestFiles, err := iceberg.ReadManifestList(bytes.NewReader(manifestListBytes))
+	require.NoError(t, err)
+	require.Len(t, manifestFiles, 1, "compaction should commit a single replacement manifest")
+	assert.Equal(t, snapshots[0].SequenceNumber+1, manifestFiles[0].SequenceNum(), "new manifest should use the committed snapshot sequence number")
+	assert.Equal(t, snapshots[0].SequenceNumber+1, manifestFiles[0].MinSequenceNum(), "compaction manifest should inherit the committed sequence number for added data")
+
+	manifestName = path.Base(manifestFiles[0].FilePath())
+	manifestBytes := readFile(t, client, metaDir, manifestName)
+	manifestEntries, err := iceberg.ReadManifest(manifestFiles[0], bytes.NewReader(manifestBytes), false)
+	require.NoError(t, err)
+	require.Len(t, manifestEntries, len(files)+1, "compaction should replace the input files with one merged output")
+
+	deletedPaths := make(map[string]struct{}, len(files))
+	addedPaths := make(map[string]struct{})
+	for _, entry := range manifestEntries {
+		switch entry.Status() {
+		case iceberg.EntryStatusADDED:
+			addedPaths[entry.DataFile().FilePath()] = struct{}{}
+			assert.Equal(t, snapshots[0].SequenceNumber+1, entry.SequenceNum(), "added entries should inherit the new snapshot sequence number")
+			fileSeqNum := entry.FileSequenceNum()
+			require.NotNil(t, fileSeqNum, "added entries should carry a file sequence number")
+			assert.Equal(t, snapshots[0].SequenceNumber+1, *fileSeqNum)
+		case iceberg.EntryStatusDELETED:
+			deletedPaths[entry.DataFile().FilePath()] = struct{}{}
+			assert.Equal(t, snapshots[0].SequenceNumber, entry.SequenceNum(), "deleted entries should preserve the original data sequence number")
+			fileSeqNum := entry.FileSequenceNum()
+			require.NotNil(t, fileSeqNum, "deleted entries should preserve file sequence numbers")
+			assert.Equal(t, snapshots[0].SequenceNumber, *fileSeqNum)
+		default:
+			t.Fatalf("unexpected manifest entry status %v for %s", entry.Status(), entry.DataFile().FilePath())
+		}
+	}
+
+	require.Len(t, addedPaths, 1, "compaction should add exactly one merged parquet file")
+	assert.Contains(t, addedPaths, path.Join("data", compacted.Name))
+	require.Len(t, deletedPaths, len(files), "compaction should delete every original small input file")
+	for _, file := range files {
+		assert.Contains(t, deletedPaths, path.Join("data", file.name))
+	}
+}
+
+func testRemoveOrphans(t *testing.T) {
+	_, client := shared.filerConn(t)
+	suffix := randomSuffix()
+	bucket := "maint-orphan-" + suffix
+	ns := "ns"
+	tbl := "tbl"
+
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+	}
+	populateTableViaFiler(t, client, shared.s3Endpoint, bucket, ns, tbl, snapshots)
+
+	ctx := context.Background()
+	tablePath := path.Join(ns, tbl)
+	tableFilerPath := path.Join(s3tables.TablesPath, bucket, tablePath)
+	dataDir := path.Join(tableFilerPath, "data")
+	metaDir := path.Join(tableFilerPath, "metadata")
+
+	// Create orphan files (old enough to be removed)
+	oldTime := time.Now().Add(-200 * time.Hour).Unix()
+	writeOrphan := func(dir, name string) {
+		resp, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
+			Directory: dir,
+			Entry: &filer_pb.Entry{
+				Name: name,
+				Attributes: &filer_pb.FuseAttributes{
+					Mtime: oldTime, Crtime: oldTime,
+					FileMode: uint32(0644), FileSize: 100,
+				},
+				Content: []byte("orphan"),
+			},
+		})
+		require.NoError(t, err, "writeOrphan(%s, %s): rpc error", dir, name)
+		require.Empty(t, resp.Error, "writeOrphan(%s, %s): resp error", dir, name)
+	}
+	writeOrphan(dataDir, "orphan-data.parquet")
+	writeOrphan(metaDir, "orphan-meta.avro")
+
+	// Create a recent orphan (should NOT be removed)
+	writeFile(t, ctx, client, dataDir, "recent-orphan.parquet", []byte("new"))
+
+	handler := icebergHandler.NewHandler(nil)
+	config := icebergHandler.Config{
+		OrphanOlderThanHours: 72,
+		MaxCommitRetries:     3,
+	}
+
+	result, _, err := handler.RemoveOrphans(ctx, client, bucket, tablePath, config)
+	require.NoError(t, err)
+	assert.Contains(t, result, "removed")
+	t.Logf("RemoveOrphans result: %s", result)
+
+	// Verify orphans were removed
+	assert.Nil(t, lookupEntry(t, client, dataDir, "orphan-data.parquet"), "old orphan data file should be deleted")
+	assert.Nil(t, lookupEntry(t, client, metaDir, "orphan-meta.avro"), "old orphan meta file should be deleted")
+
+	// Verify recent orphan was kept
+	assert.NotNil(t, lookupEntry(t, client, dataDir, "recent-orphan.parquet"), "recent orphan should be kept")
+}
+
+func testRewriteManifests(t *testing.T) {
+	_, client := shared.filerConn(t)
+	suffix := randomSuffix()
+	bucket := "maint-rewrite-" + suffix
+	ns := "ns"
+	tbl := "tbl"
+
+	// Each snapshot gets its own manifest list with one manifest. The current
+	// snapshot (latest) therefore has only 1 manifest — below the rewrite
+	// threshold. This tests the threshold check against a real filer.
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+	}
+	populateTableViaFiler(t, client, shared.s3Endpoint, bucket, ns, tbl, snapshots)
+
+	handler := icebergHandler.NewHandler(nil)
+	config := icebergHandler.Config{
+		MinManifestsToRewrite: 5, // threshold higher than 1 manifest
+		MaxCommitRetries:      3,
+	}
+
+	tablePath := path.Join(ns, tbl)
+	result, _, err := handler.RewriteManifests(context.Background(), client, bucket, tablePath, config)
+	require.NoError(t, err)
+	assert.Contains(t, result, "below threshold")
+	t.Logf("RewriteManifests result: %s", result)
+}
+
+func testFullMaintenanceCycle(t *testing.T) {
+	_, client := shared.filerConn(t)
+	suffix := randomSuffix()
+	bucket := "maint-full-" + suffix
+	ns := "ns"
+	tbl := "tbl"
+
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+		{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+	}
+	populateTableViaFiler(t, client, shared.s3Endpoint, bucket, ns, tbl, snapshots)
+
+	ctx := context.Background()
+	tablePath := path.Join(ns, tbl)
+
+	// Add an orphan
+	dataDir := path.Join(s3tables.TablesPath, bucket, tablePath, "data")
+	oldTime := time.Now().Add(-200 * time.Hour).Unix()
+	orphanResp, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
+		Directory: dataDir,
+		Entry: &filer_pb.Entry{
+			Name: "orphan.parquet",
+			Attributes: &filer_pb.FuseAttributes{
+				Mtime: oldTime, Crtime: oldTime,
+				FileMode: uint32(0644), FileSize: 100,
+			},
+			Content: []byte("orphan"),
+		},
+	})
+	require.NoError(t, err)
+	require.Empty(t, orphanResp.Error)
+
+	handler := icebergHandler.NewHandler(nil)
+
+	// Step 1: Expire snapshots
+	expireConfig := icebergHandler.Config{
+		SnapshotRetentionHours: 0, // instant expiry
+		MaxSnapshotsToKeep:     1,
+		MaxCommitRetries:       3,
+	}
+	result, _, err := handler.ExpireSnapshots(ctx, client, bucket, tablePath, expireConfig)
+	require.NoError(t, err)
+	assert.Contains(t, result, "expired")
+	t.Logf("Step 1 (expire): %s", result)
+
+	// Step 2: Remove orphans
+	orphanConfig := icebergHandler.Config{
+		OrphanOlderThanHours: 72,
+		MaxCommitRetries:     3,
+	}
+	result, _, err = handler.RemoveOrphans(ctx, client, bucket, tablePath, orphanConfig)
+	require.NoError(t, err)
+	t.Logf("Step 2 (orphans): %s", result)
+	// The orphan and the unreferenced files from expired snapshots should be gone
+	assert.Contains(t, result, "removed")
+
+	// Step 3: Verify metadata is consistent
+	entry := lookupEntry(t, client, path.Join(s3tables.TablesPath, bucket, ns), tbl)
+	require.NotNil(t, entry)
+	xattr := entry.Extended[s3tables.ExtendedKeyMetadata]
+	require.NotEmpty(t, xattr)
+
+	var internalMeta struct {
+		MetadataVersion int `json:"metadataVersion"`
+	}
+	require.NoError(t, json.Unmarshal(xattr, &internalMeta))
+	assert.Greater(t, internalMeta.MetadataVersion, 1, "metadata version should have advanced through the cycle")
+	t.Logf("Final metadata version: %d", internalMeta.MetadataVersion)
+}
@@ -23,8 +23,6 @@ func TestAdminStatusAndHealthz(t *testing.T) {
 	if err != nil {
 		t.Fatalf("create status request: %v", err)
 	}
-	statusReq.Header.Set(request_id.AmzRequestIDHeader, "test-request-id-1")
-
 	statusResp := framework.DoRequest(t, client, statusReq)
 	statusBody := framework.ReadAllAndClose(t, statusResp)

@@ -34,8 +32,8 @@ func TestAdminStatusAndHealthz(t *testing.T) {
 	if got := statusResp.Header.Get("Server"); !strings.Contains(got, "SeaweedFS Volume") {
 		t.Fatalf("expected /status Server header to contain SeaweedFS Volume, got %q", got)
 	}
-	if got := statusResp.Header.Get(request_id.AmzRequestIDHeader); got != "test-request-id-1" {
-		t.Fatalf("expected echoed request id, got %q", got)
+	if got := statusResp.Header.Get(request_id.AmzRequestIDHeader); got == "" {
+		t.Fatal("expected server-generated request id in response header")
 	}

 	var payload map[string]interface{}
@@ -49,7 +47,6 @@ func TestAdminStatusAndHealthz(t *testing.T) {
 	}

 	healthReq := mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/healthz")
-	healthReq.Header.Set(request_id.AmzRequestIDHeader, "test-request-id-2")
 	healthResp := framework.DoRequest(t, client, healthReq)
 	_ = framework.ReadAllAndClose(t, healthResp)
 	if healthResp.StatusCode != http.StatusOK {
@@ -58,8 +55,8 @@ func TestAdminStatusAndHealthz(t *testing.T) {
 	if got := healthResp.Header.Get("Server"); !strings.Contains(got, "SeaweedFS Volume") {
 		t.Fatalf("expected /healthz Server header to contain SeaweedFS Volume, got %q", got)
 	}
-	if got := healthResp.Header.Get(request_id.AmzRequestIDHeader); got != "test-request-id-2" {
-		t.Fatalf("expected /healthz echoed request id, got %q", got)
+	if got := healthResp.Header.Get(request_id.AmzRequestIDHeader); got == "" {
+		t.Fatal("expected /healthz server-generated request id in response header")
 	}

 	uiResp := framework.DoRequest(t, client, mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/ui/index.html"))
@@ -1,279 +0,0 @@
-# SeaweedFS Admin Component
-
-A modern web-based administration interface for SeaweedFS clusters built with Go, Gin, Templ, and Bootstrap.
-
-## Features
-
- **Dashboard**: Real-time cluster status and metrics
- **Master Management**: Monitor master nodes and leadership status
- **Volume Server Management**: View volume servers, capacity, and health
- **Object Store Bucket Management**: Create, delete, and manage Object Store buckets with web interface
- **S3 Tables Management**: Manage table buckets, namespaces, tables, tags, and policies via the admin UI
- **System Health**: Overall cluster health monitoring
- **Responsive Design**: Bootstrap-based UI that works on all devices
- **Authentication**: Optional user authentication with sessions
- **TLS Support**: HTTPS support for production deployments
-
-## Building
-
-### Using the Admin Makefile
-
-The admin component has its own Makefile for development and building:
-
-```bash
-# Navigate to admin directory
-cd weed/admin
-
-# View all available targets
-make help
-
-# Generate templates and build
-make build
-
-# Development mode with template watching
-make dev
-
-# Run the admin server
-make run
-
-# Clean build artifacts
-make clean
-```
-
-### Using the Root Makefile
-
-The root SeaweedFS Makefile automatically integrates the admin component:
-
-```bash
-# From the root directory
-make install          # Builds weed with admin component
-make full_install     # Full build with all tags
-make test            # Runs tests including admin component
-
-# Admin-specific targets from root
-make admin-generate  # Generate admin templates
-make admin-build     # Build admin component
-make admin-run       # Run admin server
-make admin-dev       # Development mode
-make admin-clean     # Clean admin artifacts
-```
-
-### Manual Building
-
-If you prefer to build manually:
-
-```bash
-# Install templ compiler
-go install github.com/a-h/templ/cmd/templ@latest
-
-# Generate templates
-templ generate
-
-# Build the main weed binary
-cd ../../../
-go build -o weed ./weed
-```
-
-## Development
-
-### Template Development
-
-The admin interface uses [Templ](https://templ.guide/) for type-safe HTML templates:
-
-```bash
-# Watch for template changes and auto-regenerate
-make watch
-
-# Or manually generate templates
-make generate
-
-# Format templates
-make fmt
-```
-
-### File Structure
-
-```
-weed/admin/
-├── Makefile              # Admin-specific build tasks
-├── README.md             # This file
-├── admin.go              # Main application entry point
-├── dash/                 # Server and handler logic
-│   ├── admin_server.go   # HTTP server setup
-│   ├── handler_admin.go  # Admin dashboard handlers
-│   ├── handler_auth.go   # Authentication handlers
-│   └── middleware.go     # HTTP middleware
-├── static/               # Static assets
-│   ├── css/admin.css     # Admin-specific styles
-│   └── js/admin.js       # Admin-specific JavaScript
-└── view/                 # Templates
-    ├── app/              # Application templates
-    │   ├── admin.templ   # Main dashboard template
-    │   ├── s3_buckets.templ # Object Store bucket management template
-    │   ├── s3tables_*.templ # S3 Tables management templates
-    │   └── *_templ.go    # Generated Go code
-    └── layout/           # Layout templates
-        ├── layout.templ  # Base layout template
-        └── layout_templ.go # Generated Go code
-```
-
-### Object Store Management
-
-The admin interface includes Object Store and S3 Tables management capabilities:
-
- Create/delete Object Store buckets and adjust quotas or ownership.
- Manage S3 Tables buckets, namespaces, and tables.
- Update S3 Tables policies and tags via the UI and API endpoints.
-
-## Usage
-
-### Basic Usage
-
-```bash
-# Start admin interface on default port (23646)
-weed admin
-
-# Start with custom configuration
-weed admin -port=8080 -masters="master1:9333,master2:9333"
-
-# Start with authentication
-weed admin -adminUser=admin -adminPassword=secret123
-
-# Start with HTTPS
-weed admin -port=443 -tlsCert=/path/to/cert.pem -tlsKey=/path/to/key.pem
-```
-
-### Configuration Options
-
-| Option | Default | Description |
-|--------|---------|-------------|
-| `-port` | 23646 | Admin server port |
-| `-masters` | localhost:9333 | Comma-separated master servers |
-| `-adminUser` | admin | Admin username (if auth enabled) |
-| `-adminPassword` | "" | Admin password (empty = no auth) |
-| `-tlsCert` | "" | Path to TLS certificate |
-| `-tlsKey` | "" | Path to TLS private key |
-
-### Docker Usage
-
-```bash
-# Build Docker image with admin component
-make docker-build
-
-# Run with Docker
-docker run -p 23646:23646 seaweedfs/seaweedfs:latest admin -masters=host.docker.internal:9333
-```
-
-## Development Workflow
-
-### Quick Start
-
-```bash
-# Clone and setup
-git clone <seaweedfs-repo>
-cd seaweedfs/weed/admin
-
-# Install dependencies and build
-make install-deps
-make build
-
-# Start development server
-make dev
-```
-
-### Making Changes
-
-1. **Template Changes**: Edit `.templ` files in `view/`
-   - Templates auto-regenerate in development mode
-   - Use `make generate` to manually regenerate
-
-2. **Go Code Changes**: Edit `.go` files
-   - Restart the server to see changes
-   - Use `make build` to rebuild
-
-3. **Static Assets**: Edit files in `static/`
-   - Changes are served immediately
-
-### Testing
-
-```bash
-# Run admin component tests
-make test
-
-# Run from root directory
-make admin-test
-
-# Lint code
-make lint
-
-# Format code
-make fmt
-```
-
-## Production Deployment
-
-### Security Considerations
-
-1. **Authentication**: Always set `adminPassword` in production
-2. **HTTPS**: Use TLS certificates for encrypted connections
-3. **Firewall**: Restrict admin interface access to authorized networks
-
-### Example Production Setup
-
-```bash
-# Production deployment with security
-weed admin \
-  -port=443 \
-  -masters="master1:9333,master2:9333,master3:9333" \
-  -adminUser=admin \
-  -adminPassword=your-secure-password \
-  -tlsCert=/etc/ssl/certs/admin.crt \
-  -tlsKey=/etc/ssl/private/admin.key
-```
-
-### Monitoring
-
-The admin interface provides endpoints for monitoring:
-
- `GET /health` - Health check endpoint
- `GET /metrics` - Prometheus metrics (if enabled)
- `GET /api/status` - JSON status information
-
-## Troubleshooting
-
-### Common Issues
-
-1. **Templates not found**: Run `make generate` to create template files
-2. **Build errors**: Ensure `templ` is installed with `make install-templ`
-3. **Static files not loading**: Check that `static/` directory exists and has proper files
-4. **Connection errors**: Verify master and filer addresses are correct
-
-### Debug Mode
-
-```bash
-# Enable debug logging
-weed -v=2 admin
-
-# Check generated templates
-ls -la view/app/*_templ.go view/layout/*_templ.go
-```
-
-## Contributing
-
-1. Fork the repository
-2. Create a feature branch
-3. Make your changes
-4. Run tests: `make test`
-5. Format code: `make fmt`
-6. Submit a pull request
-
-## Architecture
-
-The admin component follows a clean architecture:
-
- **Presentation Layer**: Templ templates + Bootstrap CSS
- **HTTP Layer**: Gin router with middleware
- **Business Logic**: Handler functions in `dash/` package
- **Data Layer**: Communicates with SeaweedFS masters and filers
-
-This separation makes the code maintainable and testable. 
@@ -80,6 +80,11 @@ type AccessKeyInfo struct {
 	CreatedAt time.Time `json:"created_at"`
 }

+type CreateAccessKeyRequest struct {
+	AccessKey string `json:"access_key"`
+	SecretKey string `json:"secret_key"`
+}
+
 type UpdateAccessKeyStatusRequest struct {
 	Status string `json:"status" binding:"required"`
 }
@@ -90,6 +95,7 @@ type UserDetails struct {
 	Actions     []string        `json:"actions"`
 	PolicyNames []string        `json:"policy_names"`
 	AccessKeys  []AccessKeyInfo `json:"access_keys"`
+	Groups      []string        `json:"groups"`
 }

 type FilerNode struct {
@@ -2,7 +2,9 @@ package dash

 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"sort"
 	"strings"
@@ -109,6 +111,9 @@ type AdminServer struct {
 	// Worker gRPC server
 	workerGrpcServer *WorkerGrpcServer

+	// Background goroutine lifecycle
+	bgCancel context.CancelFunc
+
 	// Collection statistics caching
 	collectionStatsCache          map[string]collectionStats
 	lastCollectionStatsUpdate     time.Time
@@ -135,8 +140,8 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 	)

 	// Start master client connection process (like shell and filer do)
-	ctx := context.Background()
-	go masterClient.KeepConnectedToMaster(ctx)
+	bgCtx, bgCancel := context.WithCancel(context.Background())
+	go masterClient.KeepConnectedToMaster(bgCtx)

 	lockManager := NewAdminLockManager(masterClient, adminLockClientName)
 	presenceLock := newAdminPresenceLock(masterClient)
@@ -157,6 +162,7 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 		icebergPort:                   icebergPort,
 		pluginLock:                    lockManager,
 		adminPresenceLock:             presenceLock,
+		bgCancel:                      bgCancel,
 	}

 	// Initialize topic retention purger
@@ -254,11 +260,93 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 	} else {
 		server.plugin = plugin
 		glog.V(0).Infof("Plugin enabled")
+		go server.monitorVacuumWorker(bgCtx)
 	}

 	return server
 }

+// vacuumToggler abstracts the master's vacuum enable/disable for testing.
+type vacuumToggler interface {
+	disableVacuum() error
+	enableVacuum() error
+}
+
+// masterVacuumToggler implements vacuumToggler via gRPC calls to the master.
+type masterVacuumToggler struct {
+	server *AdminServer
+}
+
+func (m *masterVacuumToggler) disableVacuum() error {
+	return m.server.WithMasterClient(func(client master_pb.SeaweedClient) error {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_, err := client.DisableVacuum(ctx, &master_pb.DisableVacuumRequest{ByPlugin: true})
+		return err
+	})
+}
+
+func (m *masterVacuumToggler) enableVacuum() error {
+	return m.server.WithMasterClient(func(client master_pb.SeaweedClient) error {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_, err := client.EnableVacuum(ctx, &master_pb.EnableVacuumRequest{ByPlugin: true})
+		return err
+	})
+}
+
+// syncVacuumState performs a single sync step: checks if a vacuum-capable worker
+// is present and calls disable/enable accordingly. Returns the updated state
+// and whether the call failed (for log dedup on retries).
+func syncVacuumState(hasWorker bool, previouslyActive bool, toggler vacuumToggler, retrying bool) (active bool, failed bool) {
+	if hasWorker == previouslyActive {
+		return previouslyActive, false
+	}
+	if hasWorker {
+		if !retrying {
+			glog.V(0).Infof("Vacuum plugin worker connected, disabling master automatic vacuum")
+		}
+		if err := toggler.disableVacuum(); err != nil {
+			glog.Warningf("Failed to disable vacuum on master: %v", err)
+			return false, true // retry next tick
+		}
+		return true, false
+	}
+	if !retrying {
+		glog.V(0).Infof("Vacuum plugin worker disconnected, re-enabling master automatic vacuum")
+	}
+	if err := toggler.enableVacuum(); err != nil {
+		glog.Warningf("Failed to enable vacuum on master: %v", err)
+		return true, true // retry next tick
+	}
+	return false, false
+}
+
+// monitorVacuumWorker polls the plugin registry for vacuum-capable workers and
+// disables/enables the master's automatic scheduled vacuum accordingly.
+func (s *AdminServer) monitorVacuumWorker(ctx context.Context) {
+	const pollInterval = 30 * time.Second
+	ticker := time.NewTicker(pollInterval)
+	defer ticker.Stop()
+
+	toggler := &masterVacuumToggler{server: s}
+	vacuumWorkerActive := false
+	retrying := false
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if s.plugin == nil {
+				continue
+			}
+			hasWorker := s.plugin.HasCapableWorker("vacuum")
+			vacuumWorkerActive, retrying = syncVacuumState(hasWorker, vacuumWorkerActive, toggler, retrying)
+		}
+	}
+}
+
 // loadTaskConfigurationsFromPersistence loads saved task configurations from protobuf files
 func (s *AdminServer) loadTaskConfigurationsFromPersistence() {
 	if s.configPersistence == nil || !s.configPersistence.IsConfigured() {
@@ -371,8 +459,21 @@ func (s *AdminServer) GetCredentialManager() *credential.CredentialManager {

 // InvalidateCache method moved to cluster_topology.go

-// GetS3BucketsData retrieves all Object Store buckets and aggregates total storage metrics
-func (s *AdminServer) GetS3BucketsData() (S3BucketsData, error) {
+// GetS3BucketsData retrieves Object Store buckets with pagination and sorting
+func (s *AdminServer) GetS3BucketsData(page, pageSize int, sortBy, sortOrder string) (S3BucketsData, error) {
+	if page < 1 {
+		page = 1
+	}
+	if pageSize < 1 || pageSize > 1000 {
+		pageSize = 100
+	}
+	if sortBy == "" {
+		sortBy = "name"
+	}
+	if sortOrder == "" {
+		sortOrder = "asc"
+	}
+
 	buckets, err := s.GetS3Buckets()
 	if err != nil {
 		return S3BucketsData{}, err
@@ -383,14 +484,97 @@ func (s *AdminServer) GetS3BucketsData() (S3BucketsData, error) {
 		totalSize += bucket.PhysicalSize
 	}

+	totalBuckets := len(buckets)
+
+	// Sort buckets
+	s.sortBuckets(buckets, sortBy, sortOrder)
+
+	// Calculate pagination
+	totalPages := (totalBuckets + pageSize - 1) / pageSize
+	if totalPages == 0 {
+		totalPages = 1
+	}
+	if page > totalPages {
+		page = totalPages
+	}
+
+	startIndex := (page - 1) * pageSize
+	endIndex := startIndex + pageSize
+	if startIndex >= totalBuckets {
+		buckets = []S3Bucket{}
+	} else {
+		if endIndex > totalBuckets {
+			endIndex = totalBuckets
+		}
+		buckets = buckets[startIndex:endIndex]
+	}
+
 	return S3BucketsData{
 		Buckets:      buckets,
-		TotalBuckets: len(buckets),
+		TotalBuckets: totalBuckets,
 		TotalSize:    totalSize,
 		LastUpdated:  time.Now(),
+		CurrentPage:  page,
+		TotalPages:   totalPages,
+		PageSize:     pageSize,
+		SortBy:       sortBy,
+		SortOrder:    sortOrder,
 	}, nil
 }

+// sortBuckets sorts the bucket slice in place by the given field and order
+func (s *AdminServer) sortBuckets(buckets []S3Bucket, sortBy, sortOrder string) {
+	desc := sortOrder == "desc"
+	sort.Slice(buckets, func(i, j int) bool {
+		a, b := buckets[i], buckets[j]
+		switch sortBy {
+		case "owner":
+			if a.Owner != b.Owner {
+				if desc {
+					return a.Owner > b.Owner
+				}
+				return a.Owner < b.Owner
+			}
+		case "created":
+			if !a.CreatedAt.Equal(b.CreatedAt) {
+				if desc {
+					return a.CreatedAt.After(b.CreatedAt)
+				}
+				return a.CreatedAt.Before(b.CreatedAt)
+			}
+		case "objects":
+			if a.ObjectCount != b.ObjectCount {
+				if desc {
+					return a.ObjectCount > b.ObjectCount
+				}
+				return a.ObjectCount < b.ObjectCount
+			}
+		case "logical_size":
+			if a.LogicalSize != b.LogicalSize {
+				if desc {
+					return a.LogicalSize > b.LogicalSize
+				}
+				return a.LogicalSize < b.LogicalSize
+			}
+		case "physical_size":
+			if a.PhysicalSize != b.PhysicalSize {
+				if desc {
+					return a.PhysicalSize > b.PhysicalSize
+				}
+				return a.PhysicalSize < b.PhysicalSize
+			}
+		}
+		// Tie-breaker: sort by name (also the default/primary for sortBy=="name")
+		if a.Name != b.Name {
+			if desc {
+				return a.Name > b.Name
+			}
+			return a.Name < b.Name
+		}
+		return false
+	})
+}
+
 // GetS3Buckets retrieves all Object Store buckets from the filer and collects size/object data from collections
 func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
 	var buckets []S3Bucket
@@ -406,28 +590,48 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {

 	// Now list buckets from the filer and match with collection data
 	err = s.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-		// List buckets by looking at the buckets directory
-		stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
-			Directory:          filerConfig.BucketsPath,
-			Prefix:             "",
-			StartFromFileName:  "",
-			InclusiveStartFrom: false,
-			Limit:              1000,
-		})
-		if err != nil {
-			return err
-		}
-
+		// Paginate through all buckets in the buckets directory
+		const listPageSize = 1000
+		startFrom := ""
+		var snapshotTsNs int64
 		for {
-			resp, err := stream.Recv()
+			stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
+				Directory:          filerConfig.BucketsPath,
+				Prefix:             "",
+				StartFromFileName:  startFrom,
+				InclusiveStartFrom: false,
+				Limit:              listPageSize,
+				SnapshotTsNs:       snapshotTsNs,
+			})
 			if err != nil {
-				if err.Error() == "EOF" {
-					break
-				}
 				return err
 			}

-			if resp.Entry.IsDirectory {
+			pageCount := 0
+			lastName := ""
+			for {
+				resp, err := stream.Recv()
+				if err != nil {
+					if errors.Is(err, io.EOF) {
+						break
+					}
+					return err
+				}
+
+				if snapshotTsNs == 0 && resp.SnapshotTsNs != 0 {
+					snapshotTsNs = resp.SnapshotTsNs
+				}
+
+				if resp.Entry == nil {
+					continue
+				}
+				lastName = resp.Entry.Name
+				pageCount++
+
+				if !resp.Entry.IsDirectory {
+					continue
+				}
+
 				bucketName := resp.Entry.Name
 				if strings.HasPrefix(bucketName, ".") {
 					// Skip internal/system directories from Object Store bucket listing.
@@ -480,13 +684,18 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
 					}
 				}

+				var createdAt, lastModified time.Time
+				if resp.Entry.Attributes != nil {
+					createdAt = time.Unix(resp.Entry.Attributes.Crtime, 0)
+					lastModified = time.Unix(resp.Entry.Attributes.Mtime, 0)
+				}
 				bucket := S3Bucket{
 					Name:               bucketName,
-					CreatedAt:          time.Unix(resp.Entry.Attributes.Crtime, 0),
+					CreatedAt:          createdAt,
 					LogicalSize:        logicalSize,
 					PhysicalSize:       physicalSize,
 					ObjectCount:        objectCount,
-					LastModified:       time.Unix(resp.Entry.Attributes.Mtime, 0),
+					LastModified:       lastModified,
 					Quota:              quota,
 					QuotaEnabled:       quotaEnabled,
 					VersioningStatus:   versioningStatus,
@@ -497,6 +706,12 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
 				}
 				buckets = append(buckets, bucket)
 			}
+
+			// If we received fewer entries than the page size, we've listed everything
+			if pageCount < listPageSize {
+				break
+			}
+			startFrom = lastName
 		}

 		return nil
@@ -1371,6 +1586,11 @@ func (s *AdminServer) UpdateTopicRetention(namespace, name string, enabled bool,
 func (s *AdminServer) Shutdown() {
 	glog.V(1).Infof("Shutting down admin server...")

+	// Cancel background goroutines (vacuum monitor, etc.)
+	if s.bgCancel != nil {
+		s.bgCancel()
+	}
+
 	// Stop maintenance manager
 	s.StopMaintenanceManager()
 	if s.adminPresenceLock != nil {
@@ -26,6 +26,15 @@ type S3BucketsData struct {
 	TotalBuckets int        `json:"total_buckets"`
 	TotalSize    int64      `json:"total_size"`
 	LastUpdated  time.Time  `json:"last_updated"`
+
+	// Pagination
+	CurrentPage int `json:"current_page"`
+	TotalPages  int `json:"total_pages"`
+	PageSize    int `json:"page_size"`
+
+	// Sorting
+	SortBy    string `json:"sort_by"`
+	SortOrder string `json:"sort_order"`
 }

 type CreateBucketRequest struct {
@@ -48,7 +57,7 @@ type CreateBucketRequest struct {
 func (s *AdminServer) ShowS3Buckets(w http.ResponseWriter, r *http.Request) {
 	username := UsernameFromContext(r.Context())

-	data, err := s.GetS3BucketsData()
+	data, err := s.GetS3BucketsData(1, 100, "name", "asc")
 	if err != nil {
 		writeJSONError(w, http.StatusInternalServerError, "Failed to get Object Store buckets: "+err.Error())
 		return
@@ -0,0 +1,250 @@
+package dash
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/credential"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+)
+
+// cloneGroup creates a deep copy of an iam_pb.Group to avoid mutating stored state.
+func cloneGroup(g *iam_pb.Group) *iam_pb.Group {
+	clone := &iam_pb.Group{
+		Name:     g.Name,
+		Disabled: g.Disabled,
+	}
+	if g.Members != nil {
+		clone.Members = make([]string, len(g.Members))
+		copy(clone.Members, g.Members)
+	}
+	if g.PolicyNames != nil {
+		clone.PolicyNames = make([]string, len(g.PolicyNames))
+		copy(clone.PolicyNames, g.PolicyNames)
+	}
+	return clone
+}
+
+func (s *AdminServer) GetGroups(ctx context.Context) ([]GroupData, error) {
+	if s.credentialManager == nil {
+		return nil, fmt.Errorf("credential manager not available")
+	}
+
+	groupNames, err := s.credentialManager.ListGroups(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list groups: %w", err)
+	}
+
+	var groups []GroupData
+	for _, name := range groupNames {
+		g, err := s.credentialManager.GetGroup(ctx, name)
+		if err != nil {
+			if errors.Is(err, credential.ErrGroupNotFound) {
+				glog.V(1).Infof("Group %s listed but not found, skipping", name)
+				continue
+			}
+			return nil, fmt.Errorf("failed to get group %s: %w", name, err)
+		}
+		status := "enabled"
+		if g.Disabled {
+			status = "disabled"
+		}
+		groups = append(groups, GroupData{
+			Name:        g.Name,
+			MemberCount: len(g.Members),
+			PolicyCount: len(g.PolicyNames),
+			Status:      status,
+			Members:     g.Members,
+			PolicyNames: g.PolicyNames,
+		})
+	}
+	return groups, nil
+}
+
+func (s *AdminServer) GetGroupDetails(ctx context.Context, name string) (*GroupData, error) {
+	if s.credentialManager == nil {
+		return nil, fmt.Errorf("credential manager not available")
+	}
+
+	g, err := s.credentialManager.GetGroup(ctx, name)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get group: %w", err)
+	}
+	status := "enabled"
+	if g.Disabled {
+		status = "disabled"
+	}
+	return &GroupData{
+		Name:        g.Name,
+		MemberCount: len(g.Members),
+		PolicyCount: len(g.PolicyNames),
+		Status:      status,
+		Members:     g.Members,
+		PolicyNames: g.PolicyNames,
+	}, nil
+}
+
+func (s *AdminServer) CreateGroup(ctx context.Context, name string) (*GroupData, error) {
+	if s.credentialManager == nil {
+		return nil, fmt.Errorf("credential manager not available")
+	}
+
+	group := &iam_pb.Group{Name: name}
+	if err := s.credentialManager.CreateGroup(ctx, group); err != nil {
+		return nil, fmt.Errorf("failed to create group: %w", err)
+	}
+	glog.V(1).Infof("Created group %s", group.Name)
+	return &GroupData{
+		Name:   group.Name,
+		Status: "enabled",
+	}, nil
+}
+
+func (s *AdminServer) DeleteGroup(ctx context.Context, name string) error {
+	if s.credentialManager == nil {
+		return fmt.Errorf("credential manager not available")
+	}
+	// Check for members and attached policies before deleting (same guards as IAM handlers)
+	g, err := s.credentialManager.GetGroup(ctx, name)
+	if err != nil {
+		return fmt.Errorf("failed to get group: %w", err)
+	}
+	if len(g.Members) > 0 {
+		return fmt.Errorf("cannot delete group %s: group has %d member(s): %w", name, len(g.Members), credential.ErrGroupNotEmpty)
+	}
+	if len(g.PolicyNames) > 0 {
+		return fmt.Errorf("cannot delete group %s: group has %d attached policy(ies): %w", name, len(g.PolicyNames), credential.ErrGroupNotEmpty)
+	}
+	if err := s.credentialManager.DeleteGroup(ctx, name); err != nil {
+		return fmt.Errorf("failed to delete group: %w", err)
+	}
+	glog.V(1).Infof("Deleted group %s", name)
+	return nil
+}
+
+func (s *AdminServer) AddGroupMember(ctx context.Context, groupName, username string) error {
+	if s.credentialManager == nil {
+		return fmt.Errorf("credential manager not available")
+	}
+	g, err := s.credentialManager.GetGroup(ctx, groupName)
+	if err != nil {
+		return fmt.Errorf("failed to get group: %w", err)
+	}
+	g = cloneGroup(g)
+	if _, err := s.credentialManager.GetUser(ctx, username); err != nil {
+		return fmt.Errorf("user %s not found: %w", username, err)
+	}
+	for _, m := range g.Members {
+		if m == username {
+			return nil // already a member
+		}
+	}
+	g.Members = append(g.Members, username)
+	if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
+		return fmt.Errorf("failed to update group: %w", err)
+	}
+	glog.V(1).Infof("Added user %s to group %s", username, groupName)
+	return nil
+}
+
+func (s *AdminServer) RemoveGroupMember(ctx context.Context, groupName, username string) error {
+	if s.credentialManager == nil {
+		return fmt.Errorf("credential manager not available")
+	}
+	g, err := s.credentialManager.GetGroup(ctx, groupName)
+	if err != nil {
+		return fmt.Errorf("failed to get group: %w", err)
+	}
+	g = cloneGroup(g)
+	found := false
+	var newMembers []string
+	for _, m := range g.Members {
+		if m == username {
+			found = true
+		} else {
+			newMembers = append(newMembers, m)
+		}
+	}
+	if !found {
+		return fmt.Errorf("user %s is not a member of group %s: %w", username, groupName, credential.ErrUserNotInGroup)
+	}
+	g.Members = newMembers
+	if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
+		return fmt.Errorf("failed to update group: %w", err)
+	}
+	glog.V(1).Infof("Removed user %s from group %s", username, groupName)
+	return nil
+}
+
+func (s *AdminServer) AttachGroupPolicy(ctx context.Context, groupName, policyName string) error {
+	if s.credentialManager == nil {
+		return fmt.Errorf("credential manager not available")
+	}
+	g, err := s.credentialManager.GetGroup(ctx, groupName)
+	if err != nil {
+		return fmt.Errorf("failed to get group: %w", err)
+	}
+	g = cloneGroup(g)
+	if _, err := s.credentialManager.GetPolicy(ctx, policyName); err != nil {
+		return fmt.Errorf("policy %s not found: %w", policyName, err)
+	}
+	for _, p := range g.PolicyNames {
+		if p == policyName {
+			return nil // already attached
+		}
+	}
+	g.PolicyNames = append(g.PolicyNames, policyName)
+	if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
+		return fmt.Errorf("failed to update group: %w", err)
+	}
+	glog.V(1).Infof("Attached policy %s to group %s", policyName, groupName)
+	return nil
+}
+
+func (s *AdminServer) DetachGroupPolicy(ctx context.Context, groupName, policyName string) error {
+	if s.credentialManager == nil {
+		return fmt.Errorf("credential manager not available")
+	}
+	g, err := s.credentialManager.GetGroup(ctx, groupName)
+	if err != nil {
+		return fmt.Errorf("failed to get group: %w", err)
+	}
+	g = cloneGroup(g)
+	found := false
+	var newPolicies []string
+	for _, p := range g.PolicyNames {
+		if p == policyName {
+			found = true
+		} else {
+			newPolicies = append(newPolicies, p)
+		}
+	}
+	if !found {
+		return fmt.Errorf("policy %s is not attached to group %s: %w", policyName, groupName, credential.ErrPolicyNotAttached)
+	}
+	g.PolicyNames = newPolicies
+	if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
+		return fmt.Errorf("failed to update group: %w", err)
+	}
+	glog.V(1).Infof("Detached policy %s from group %s", policyName, groupName)
+	return nil
+}
+
+func (s *AdminServer) SetGroupStatus(ctx context.Context, groupName string, enabled bool) error {
+	if s.credentialManager == nil {
+		return fmt.Errorf("credential manager not available")
+	}
+	g, err := s.credentialManager.GetGroup(ctx, groupName)
+	if err != nil {
+		return fmt.Errorf("failed to get group: %w", err)
+	}
+	g = cloneGroup(g)
+	g.Disabled = !enabled
+	if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
+		return fmt.Errorf("failed to update group: %w", err)
+	}
+	glog.V(1).Infof("Set group %s status to enabled=%v", groupName, enabled)
+	return nil
+}
@@ -324,7 +324,7 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
 			}

 			// Only process directories that are versions (start with "v")
-			if versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
+			if versionResp.Entry != nil && versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
 				versionDir := filepath.Join(topicDir, versionResp.Entry.Name)

 				// List all partition directories under the version directory (e.g., 0315-0630)
@@ -352,7 +352,7 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
 					}

 					// Only process directories that are partitions (format: NNNN-NNNN)
-					if partitionResp.Entry.IsDirectory {
+					if partitionResp.Entry != nil && partitionResp.Entry.IsDirectory {
 						// Parse partition range to get partition start ID (e.g., "0315-0630" -> 315)
 						var partitionStart, partitionStop int32
 						if n, err := fmt.Sscanf(partitionResp.Entry.Name, "%04d-%04d", &partitionStart, &partitionStop); n != 2 || err != nil {
@@ -387,11 +387,11 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
 							}

 							// Only process .offset files
-							if !offsetResp.Entry.IsDirectory && strings.HasSuffix(offsetResp.Entry.Name, ".offset") {
+							if offsetResp.Entry != nil && !offsetResp.Entry.IsDirectory && strings.HasSuffix(offsetResp.Entry.Name, ".offset") {
 								consumerGroup := strings.TrimSuffix(offsetResp.Entry.Name, ".offset")

 								// Read the offset value from the file
-								offsetData, err := filer.ReadInsideFiler(client, partitionDir, offsetResp.Entry.Name)
+								offsetData, err := filer.ReadInsideFiler(context.Background(), client, partitionDir, offsetResp.Entry.Name)
 								if err != nil {
 									glog.Warningf("Failed to read offset file %s: %v", offsetResp.Entry.Name, err)
 									continue
@@ -401,7 +401,10 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
 									offset := int64(util.BytesToUint64(offsetData))

 									// Get the file modification time
-									lastUpdated := time.Unix(offsetResp.Entry.Attributes.Mtime, 0)
+									var lastUpdated time.Time
+									if offsetResp.Entry.Attributes != nil {
+										lastUpdated = time.Unix(offsetResp.Entry.Attributes.Mtime, 0)
+									}

 									offsets = append(offsets, ConsumerGroupOffsetInfo{
 										ConsumerGroup: consumerGroup,
@@ -235,53 +235,6 @@ func (s *AdminServer) GetPluginSchedulerStatusAPI(w http.ResponseWriter, r *http
 	writeJSON(w, http.StatusOK, response)
 }

-// GetPluginSchedulerConfigAPI returns scheduler configuration.
-func (s *AdminServer) GetPluginSchedulerConfigAPI(w http.ResponseWriter, r *http.Request) {
-	pluginSvc := s.GetPlugin()
-	if pluginSvc == nil {
-		writeJSONError(w, http.StatusNotFound, "plugin is not enabled")
-		return
-	}
-
-	writeJSON(w, http.StatusOK, pluginSvc.GetSchedulerConfig())
-}
-
-// UpdatePluginSchedulerConfigAPI updates scheduler configuration.
-func (s *AdminServer) UpdatePluginSchedulerConfigAPI(w http.ResponseWriter, r *http.Request) {
-	pluginSvc := s.GetPlugin()
-	if pluginSvc == nil {
-		writeJSONError(w, http.StatusNotFound, "plugin is not enabled")
-		return
-	}
-
-	var req struct {
-		IdleSleepSeconds *int32 `json:"idle_sleep_seconds"`
-	}
-
-	if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
-		if errors.Is(err, io.EOF) {
-			writeJSONError(w, http.StatusBadRequest, "request body is required")
-			return
-		}
-		writeJSONError(w, http.StatusBadRequest, "invalid request body: "+err.Error())
-		return
-	}
-	if req.IdleSleepSeconds == nil {
-		writeJSONError(w, http.StatusBadRequest, "idle_sleep_seconds is required")
-		return
-	}
-
-	updated, err := pluginSvc.UpdateSchedulerConfig(plugin.SchedulerConfig{
-		IdleSleepSeconds: *req.IdleSleepSeconds,
-	})
-	if err != nil {
-		writeJSONError(w, http.StatusInternalServerError, err.Error())
-		return
-	}
-
-	writeJSON(w, http.StatusOK, updated)
-}
-
 // RequestPluginJobTypeSchemaAPI asks a worker for one job type schema.
 func (s *AdminServer) RequestPluginJobTypeSchemaAPI(w http.ResponseWriter, r *http.Request) {
 	jobType := strings.TrimSpace(mux.Vars(r)["jobType"])
@@ -151,17 +151,21 @@ func (p *TopicRetentionPurger) purgeTopicData(topicRetention TopicRetentionConfi
 			}

 			// Only process directories that are versions (start with "v")
-			if versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
+			if versionResp.Entry != nil && versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
 				versionTime, err := p.parseVersionTime(versionResp.Entry.Name)
 				if err != nil {
 					glog.Warningf("Failed to parse version time from %s: %v", versionResp.Entry.Name, err)
 					continue
 				}

+				var modTime time.Time
+				if versionResp.Entry.Attributes != nil {
+					modTime = time.Unix(versionResp.Entry.Attributes.Mtime, 0)
+				}
 				versionDirs = append(versionDirs, VersionDirInfo{
 					Name:        versionResp.Entry.Name,
 					VersionTime: versionTime,
-					ModTime:     time.Unix(versionResp.Entry.Attributes.Mtime, 0),
+					ModTime:     modTime,
 				})
 			}
 		}
@@ -260,6 +264,9 @@ func (p *TopicRetentionPurger) deleteDirectoryRecursively(client filer_pb.Seawee
 			return fmt.Errorf("failed to receive entries: %w", err)
 		}

+		if resp.Entry == nil {
+			continue
+		}
 		entryPath := filepath.Join(dirPath, resp.Entry.Name)

 		if resp.Entry.IsDirectory {
@@ -589,6 +589,30 @@ type UpdateServiceAccountRequest struct {
 	Expiration  string `json:"expiration,omitempty"`
 }

+// Group management structures
+type GroupData struct {
+	Name        string   `json:"name"`
+	MemberCount int      `json:"member_count"`
+	PolicyCount int      `json:"policy_count"`
+	Status      string   `json:"status"` // "enabled" or "disabled"
+	Members     []string `json:"members"`
+	PolicyNames []string `json:"policy_names"`
+}
+
+type GroupsPageData struct {
+	Username          string      `json:"username"`
+	Groups            []GroupData `json:"groups"`
+	TotalGroups       int         `json:"total_groups"`
+	ActiveGroups      int         `json:"active_groups"`
+	AvailableUsers    []string    `json:"available_users"`
+	AvailablePolicies []string    `json:"available_policies"`
+	LastUpdated       time.Time   `json:"last_updated"`
+}
+
+type CreateGroupRequest struct {
+	Name string `json:"name"`
+}
+
 // STS Configuration display types
 type STSConfigData struct {
 	Enabled       bool      `json:"enabled"`
@@ -4,13 +4,21 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/base64"
+	"errors"
 	"fmt"
+	"strings"
 	"time"

 	"github.com/seaweedfs/seaweedfs/weed/credential"
 	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
 )

+var (
+	ErrAccessKeyInUse = errors.New("access key already in use")
+	ErrUserNotFound   = errors.New("user not found")
+	ErrInvalidInput   = errors.New("invalid input")
+)
+
 // CreateObjectStoreUser creates a new user using the credential manager
 func (s *AdminServer) CreateObjectStoreUser(req CreateUserRequest) (*ObjectStoreUser, error) {
 	if s.credentialManager == nil {
@@ -187,6 +195,24 @@ func (s *AdminServer) GetObjectStoreUserDetails(username string) (*UserDetails,
 		details.Email = identity.Account.EmailAddress
 	}

+	// Look up groups the user belongs to
+	groupNames, err := s.credentialManager.ListGroups(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list groups: %w", err)
+	}
+	for _, gName := range groupNames {
+		g, err := s.credentialManager.GetGroup(ctx, gName)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get group %s: %w", gName, err)
+		}
+		for _, member := range g.Members {
+			if member == username {
+				details.Groups = append(details.Groups, gName)
+				break
+			}
+		}
+	}
+
 	// Convert credentials to access key info
 	for _, cred := range identity.Credentials {
 		details.AccessKeys = append(details.AccessKeys, AccessKeyInfo{
@@ -201,7 +227,7 @@ func (s *AdminServer) GetObjectStoreUserDetails(username string) (*UserDetails,
 }

 // CreateAccessKey creates a new access key for a user
-func (s *AdminServer) CreateAccessKey(username string) (*AccessKeyInfo, error) {
+func (s *AdminServer) CreateAccessKey(username string, req *CreateAccessKeyRequest) (*AccessKeyInfo, error) {
 	if s.credentialManager == nil {
 		return nil, fmt.Errorf("credential manager not available")
 	}
@@ -212,14 +238,41 @@ func (s *AdminServer) CreateAccessKey(username string) (*AccessKeyInfo, error) {
 	_, err := s.credentialManager.GetUser(ctx, username)
 	if err != nil {
 		if err == credential.ErrUserNotFound {
-			return nil, fmt.Errorf("user %s not found", username)
+			return nil, fmt.Errorf("user %s: %w", username, ErrUserNotFound)
 		}
 		return nil, fmt.Errorf("failed to get user: %w", err)
 	}

-	// Generate new access key
-	accessKey := generateAccessKey()
-	secretKey := generateSecretKey()
+	if req == nil {
+		req = &CreateAccessKeyRequest{}
+	}
+
+	// Validate provided keys
+	if req.AccessKey != "" && (len(req.AccessKey) < 4 || len(req.AccessKey) > 128) {
+		return nil, fmt.Errorf("access key must be between 4 and 128 characters: %w", ErrInvalidInput)
+	}
+	if req.SecretKey != "" && (len(req.SecretKey) < 8 || len(req.SecretKey) > 128) {
+		return nil, fmt.Errorf("secret key must be between 8 and 128 characters: %w", ErrInvalidInput)
+	}
+
+	// Use provided keys or generate new ones
+	accessKey := req.AccessKey
+	if accessKey == "" {
+		accessKey = generateAccessKey()
+	}
+	secretKey := req.SecretKey
+	if secretKey == "" {
+		secretKey = generateSecretKey()
+	}
+
+	// Verify access key is globally unique
+	existingUser, err := s.credentialManager.GetUserByAccessKey(ctx, accessKey)
+	if existingUser != nil {
+		return nil, ErrAccessKeyInUse
+	}
+	if err != nil && !errors.Is(err, credential.ErrAccessKeyNotFound) && !isNotFoundError(err) {
+		return nil, fmt.Errorf("failed to check access key uniqueness: %w", err)
+	}

 	credential := &iam_pb.Credential{
 		AccessKey: accessKey,
@@ -364,6 +417,12 @@ func (s *AdminServer) UpdateUserPolicies(username string, actions []string) erro
 	return nil
 }

+// isNotFoundError checks for "not found" in the error message as a fallback
+// for stores (e.g. gRPC) that don't return the credential.ErrAccessKeyNotFound sentinel.
+func isNotFoundError(err error) bool {
+	return err != nil && strings.Contains(strings.ToLower(err.Error()), "not found")
+}
+
 // Helper functions for generating keys and IDs
 func generateAccessKey() string {
 	// Generate 20-character access key (AWS standard)
@@ -0,0 +1,116 @@
+package dash
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+
+	adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin"
+)
+
+// TestVacuumMonitorStateTransitions verifies the state transition logic that
+// drives the vacuum monitor goroutine: the plugin registry correctly reports
+// whether a vacuum-capable worker is present or absent.
+func TestVacuumMonitorStateTransitions(t *testing.T) {
+	t.Parallel()
+
+	opts := adminplugin.Options{
+		DataDir: "",
+	}
+	p, err := adminplugin.New(opts)
+	if err != nil {
+		t.Fatalf("failed to create plugin: %v", err)
+	}
+	defer p.Shutdown()
+
+	// Initially no workers => no capable worker.
+	if p.HasCapableWorker("vacuum") {
+		t.Fatalf("expected no capable worker initially")
+	}
+
+	// Simulate state transition false -> true: worker connects.
+	p.WorkerConnectForTest(&plugin_pb.WorkerHello{
+		WorkerId: "vacuum-worker-1",
+		Capabilities: []*plugin_pb.JobTypeCapability{
+			{JobType: "vacuum", CanDetect: true, CanExecute: true},
+		},
+	})
+
+	if !p.HasCapableWorker("vacuum") {
+		t.Fatalf("expected capable worker after connect")
+	}
+
+	// Simulate state transition true -> false: worker disconnects.
+	p.WorkerDisconnectForTest("vacuum-worker-1")
+
+	// Give registry removal a moment (it's synchronous, but be safe).
+	time.Sleep(10 * time.Millisecond)
+
+	if p.HasCapableWorker("vacuum") {
+		t.Fatalf("expected no capable worker after disconnect")
+	}
+}
+
+// fakeToggler records disable/enable calls for testing.
+type fakeToggler struct {
+	disableCalls int
+	enableCalls  int
+}
+
+func (f *fakeToggler) disableVacuum() error {
+	f.disableCalls++
+	return nil
+}
+
+func (f *fakeToggler) enableVacuum() error {
+	f.enableCalls++
+	return nil
+}
+
+func TestSyncVacuumState(t *testing.T) {
+	t.Parallel()
+
+	t.Run("no change when state matches", func(t *testing.T) {
+		tog := &fakeToggler{}
+		result, failed := syncVacuumState(false, false, tog, false)
+		if result != false || failed {
+			t.Error("expected false, no failure")
+		}
+		result, failed = syncVacuumState(true, true, tog, false)
+		if result != true || failed {
+			t.Error("expected true, no failure")
+		}
+		if tog.disableCalls != 0 || tog.enableCalls != 0 {
+			t.Errorf("expected no calls, got disable=%d enable=%d", tog.disableCalls, tog.enableCalls)
+		}
+	})
+
+	t.Run("worker connects triggers disable", func(t *testing.T) {
+		tog := &fakeToggler{}
+		result, failed := syncVacuumState(true, false, tog, false)
+		if result != true || failed {
+			t.Error("expected true after disable, no failure")
+		}
+		if tog.disableCalls != 1 {
+			t.Errorf("expected 1 disable call, got %d", tog.disableCalls)
+		}
+		if tog.enableCalls != 0 {
+			t.Errorf("expected 0 enable calls, got %d", tog.enableCalls)
+		}
+	})
+
+	t.Run("worker disconnects triggers enable", func(t *testing.T) {
+		tog := &fakeToggler{}
+		result, failed := syncVacuumState(false, true, tog, false)
+		if result != false || failed {
+			t.Error("expected false after enable, no failure")
+		}
+		if tog.enableCalls != 1 {
+			t.Errorf("expected 1 enable call, got %d", tog.enableCalls)
+		}
+		if tog.disableCalls != 0 {
+			t.Errorf("expected 0 disable calls, got %d", tog.disableCalls)
+		}
+	})
+}
@@ -457,6 +457,7 @@ func (s *AdminServer) GetClusterVolumeServers() (*ClusterVolumeServersData, erro

 						// Process disk information
 						for _, diskInfo := range node.DiskInfos {
+							vs.MaxVolumes += int(diskInfo.MaxVolumeCount)
 							vs.DiskCapacity += int64(diskInfo.MaxVolumeCount) * int64(volumeSizeLimitMB) * 1024 * 1024 // Use actual volume size limit

 							// Count regular volumes and calculate disk usage
@@ -3,6 +3,7 @@ package handlers
 import (
 	"net/http"
 	"net/url"
+	"strconv"
 	"time"

 	"github.com/gorilla/mux"
@@ -28,6 +29,7 @@ type AdminHandlers struct {
 	pluginHandlers         *PluginHandlers
 	mqHandlers             *MessageQueueHandlers
 	serviceAccountHandlers *ServiceAccountHandlers
+	groupHandlers          *GroupHandlers
 }

 // NewAdminHandlers creates a new instance of AdminHandlers
@@ -40,6 +42,7 @@ func NewAdminHandlers(adminServer *dash.AdminServer, store sessions.Store) *Admi
 	pluginHandlers := NewPluginHandlers(adminServer)
 	mqHandlers := NewMessageQueueHandlers(adminServer)
 	serviceAccountHandlers := NewServiceAccountHandlers(adminServer)
+	groupHandlers := NewGroupHandlers(adminServer)
 	return &AdminHandlers{
 		adminServer:            adminServer,
 		sessionStore:           store,
@@ -51,6 +54,7 @@ func NewAdminHandlers(adminServer *dash.AdminServer, store sessions.Store) *Admi
 		pluginHandlers:         pluginHandlers,
 		mqHandlers:             mqHandlers,
 		serviceAccountHandlers: serviceAccountHandlers,
+		groupHandlers:          groupHandlers,
 	}
 }

@@ -104,6 +108,7 @@ func (h *AdminHandlers) registerUIRoutes(r *mux.Router) {
 	r.HandleFunc("/object-store/buckets/{bucket}", h.ShowBucketDetails).Methods(http.MethodGet)
 	r.HandleFunc("/object-store/users", h.userHandlers.ShowObjectStoreUsers).Methods(http.MethodGet)
 	r.HandleFunc("/object-store/policies", h.policyHandlers.ShowPolicies).Methods(http.MethodGet)
+	r.HandleFunc("/object-store/groups", h.groupHandlers.ShowGroups).Methods(http.MethodGet)
 	r.HandleFunc("/object-store/service-accounts", h.serviceAccountHandlers.ShowServiceAccounts).Methods(http.MethodGet)
 	r.HandleFunc("/object-store/s3tables/buckets", h.ShowS3TablesBuckets).Methods(http.MethodGet)
 	r.HandleFunc("/object-store/s3tables/buckets/{bucket}/namespaces", h.ShowS3TablesNamespaces).Methods(http.MethodGet)
@@ -185,6 +190,19 @@ func (h *AdminHandlers) registerAPIRoutes(api *mux.Router, enforceWrite bool) {
 	saApi.Handle("/{id}", wrapWrite(h.serviceAccountHandlers.UpdateServiceAccount)).Methods(http.MethodPut)
 	saApi.Handle("/{id}", wrapWrite(h.serviceAccountHandlers.DeleteServiceAccount)).Methods(http.MethodDelete)

+	groupsApi := api.PathPrefix("/groups").Subrouter()
+	groupsApi.HandleFunc("", h.groupHandlers.GetGroups).Methods(http.MethodGet)
+	groupsApi.Handle("", wrapWrite(h.groupHandlers.CreateGroup)).Methods(http.MethodPost)
+	groupsApi.HandleFunc("/{name}", h.groupHandlers.GetGroupDetails).Methods(http.MethodGet)
+	groupsApi.Handle("/{name}", wrapWrite(h.groupHandlers.DeleteGroup)).Methods(http.MethodDelete)
+	groupsApi.Handle("/{name}/status", wrapWrite(h.groupHandlers.SetGroupStatus)).Methods(http.MethodPut)
+	groupsApi.HandleFunc("/{name}/members", h.groupHandlers.GetGroupMembers).Methods(http.MethodGet)
+	groupsApi.Handle("/{name}/members", wrapWrite(h.groupHandlers.AddGroupMember)).Methods(http.MethodPost)
+	groupsApi.Handle("/{name}/members/{username}", wrapWrite(h.groupHandlers.RemoveGroupMember)).Methods(http.MethodDelete)
+	groupsApi.HandleFunc("/{name}/policies", h.groupHandlers.GetGroupPolicies).Methods(http.MethodGet)
+	groupsApi.Handle("/{name}/policies", wrapWrite(h.groupHandlers.AttachGroupPolicy)).Methods(http.MethodPost)
+	groupsApi.Handle("/{name}/policies/{policyName}", wrapWrite(h.groupHandlers.DetachGroupPolicy)).Methods(http.MethodDelete)
+
 	policyApi := api.PathPrefix("/object-store/policies").Subrouter()
 	policyApi.HandleFunc("", h.policyHandlers.GetPolicies).Methods(http.MethodGet)
 	policyApi.Handle("", wrapWrite(h.policyHandlers.CreatePolicy)).Methods(http.MethodPost)
@@ -229,8 +247,6 @@ func (h *AdminHandlers) registerAPIRoutes(api *mux.Router, enforceWrite bool) {
 	pluginApi.HandleFunc("/status", h.adminServer.GetPluginStatusAPI).Methods(http.MethodGet)
 	pluginApi.HandleFunc("/workers", h.adminServer.GetPluginWorkersAPI).Methods(http.MethodGet)
 	pluginApi.HandleFunc("/job-types", h.adminServer.GetPluginJobTypesAPI).Methods(http.MethodGet)
-	pluginApi.HandleFunc("/scheduler-config", h.adminServer.GetPluginSchedulerConfigAPI).Methods(http.MethodGet)
-	pluginApi.Handle("/scheduler-config", wrapWrite(h.adminServer.UpdatePluginSchedulerConfigAPI)).Methods(http.MethodPut)
 	pluginApi.HandleFunc("/jobs", h.adminServer.GetPluginJobsAPI).Methods(http.MethodGet)
 	pluginApi.HandleFunc("/jobs/{jobId}", h.adminServer.GetPluginJobAPI).Methods(http.MethodGet)
 	pluginApi.HandleFunc("/jobs/{jobId}/detail", h.adminServer.GetPluginJobDetailAPI).Methods(http.MethodGet)
@@ -278,8 +294,26 @@ func (h *AdminHandlers) ShowDashboard(w http.ResponseWriter, r *http.Request) {

 // ShowS3Buckets renders the Object Store buckets management page
 func (h *AdminHandlers) ShowS3Buckets(w http.ResponseWriter, r *http.Request) {
-	// Get Object Store buckets data from the server
-	s3Data := h.getS3BucketsData(r)
+	// Get pagination and sorting parameters from query string
+	page := 1
+	if p := r.URL.Query().Get("page"); p != "" {
+		if parsed, err := strconv.Atoi(p); err == nil && parsed > 0 {
+			page = parsed
+		}
+	}
+
+	pageSize := 100
+	if ps := r.URL.Query().Get("pageSize"); ps != "" {
+		if parsed, err := strconv.Atoi(ps); err == nil && parsed > 0 && parsed <= 1000 {
+			pageSize = parsed
+		}
+	}
+
+	sortBy := defaultQuery(r.URL.Query().Get("sortBy"), "name")
+	sortOrder := defaultQuery(r.URL.Query().Get("sortOrder"), "asc")
+
+	// Get Object Store buckets data with pagination
+	s3Data := h.getS3BucketsData(r, page, pageSize, sortBy, sortOrder)
 	username := h.getUsername(r)

 	// Render HTML template
@@ -446,15 +480,15 @@ func (h *AdminHandlers) ShowBucketDetails(w http.ResponseWriter, r *http.Request
 	writeJSON(w, http.StatusOK, details)
 }

-// getS3BucketsData retrieves Object Store buckets data from the server
-func (h *AdminHandlers) getS3BucketsData(r *http.Request) dash.S3BucketsData {
+// getS3BucketsData retrieves Object Store buckets data from the server with pagination
+func (h *AdminHandlers) getS3BucketsData(r *http.Request, page, pageSize int, sortBy, sortOrder string) dash.S3BucketsData {
 	username := dash.UsernameFromContext(r.Context())
 	if username == "" {
 		username = "admin"
 	}

 	// Get Object Store buckets data
-	data, err := h.adminServer.GetS3BucketsData()
+	data, err := h.adminServer.GetS3BucketsData(page, pageSize, sortBy, sortOrder)
 	if err != nil {
 		// Return empty data on error
 		return dash.S3BucketsData{
@@ -463,6 +497,11 @@ func (h *AdminHandlers) getS3BucketsData(r *http.Request) dash.S3BucketsData {
 			TotalBuckets: 0,
 			TotalSize:    0,
 			LastUpdated:  time.Now(),
+			CurrentPage:  1,
+			TotalPages:   1,
+			PageSize:     pageSize,
+			SortBy:       sortBy,
+			SortOrder:    sortOrder,
 		}
 	}

@@ -0,0 +1,271 @@
+package handlers
+
+import (
+	"bytes"
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/admin/dash"
+	"github.com/seaweedfs/seaweedfs/weed/admin/view/app"
+	"github.com/seaweedfs/seaweedfs/weed/admin/view/layout"
+	"github.com/seaweedfs/seaweedfs/weed/credential"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+func groupErrorToHTTPStatus(err error) int {
+	if errors.Is(err, credential.ErrGroupNotFound) {
+		return http.StatusNotFound
+	}
+	if errors.Is(err, credential.ErrGroupAlreadyExists) {
+		return http.StatusConflict
+	}
+	if errors.Is(err, credential.ErrUserNotInGroup) {
+		return http.StatusBadRequest
+	}
+	if errors.Is(err, credential.ErrPolicyNotAttached) {
+		return http.StatusBadRequest
+	}
+	if errors.Is(err, credential.ErrUserNotFound) {
+		return http.StatusNotFound
+	}
+	if errors.Is(err, credential.ErrPolicyNotFound) {
+		return http.StatusNotFound
+	}
+	if errors.Is(err, credential.ErrGroupNotEmpty) {
+		return http.StatusConflict
+	}
+	return http.StatusInternalServerError
+}
+
+type GroupHandlers struct {
+	adminServer *dash.AdminServer
+}
+
+func NewGroupHandlers(adminServer *dash.AdminServer) *GroupHandlers {
+	return &GroupHandlers{adminServer: adminServer}
+}
+
+func (h *GroupHandlers) ShowGroups(w http.ResponseWriter, r *http.Request) {
+	data, err := h.getGroupsPageData(r)
+	if err != nil {
+		glog.Errorf("Failed to get groups data: %v", err)
+		writeJSONError(w, http.StatusInternalServerError, "Failed to load groups: "+err.Error())
+		return
+	}
+
+	var buf bytes.Buffer
+	component := app.Groups(data)
+	viewCtx := layout.NewViewContext(r, dash.UsernameFromContext(r.Context()), dash.CSRFTokenFromContext(r.Context()))
+	layoutComponent := layout.Layout(viewCtx, component)
+	if err := layoutComponent.Render(r.Context(), &buf); err != nil {
+		glog.Errorf("Failed to render groups template: %v", err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "text/html")
+	_, _ = w.Write(buf.Bytes())
+}
+
+func (h *GroupHandlers) GetGroups(w http.ResponseWriter, r *http.Request) {
+	groups, err := h.adminServer.GetGroups(r.Context())
+	if err != nil {
+		glog.Errorf("Failed to get groups: %v", err)
+		writeJSONError(w, http.StatusInternalServerError, "Failed to get groups")
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"groups": groups})
+}
+
+func (h *GroupHandlers) CreateGroup(w http.ResponseWriter, r *http.Request) {
+	var req dash.CreateGroupRequest
+	if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
+		writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
+		return
+	}
+	if req.Name == "" {
+		writeJSONError(w, http.StatusBadRequest, "Group name is required")
+		return
+	}
+	group, err := h.adminServer.CreateGroup(r.Context(), req.Name)
+	if err != nil {
+		glog.Errorf("Failed to create group: %v", err)
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to create group: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, group)
+}
+
+func (h *GroupHandlers) GetGroupDetails(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	group, err := h.adminServer.GetGroupDetails(r.Context(), name)
+	if err != nil {
+		glog.Errorf("Failed to get group details: %v", err)
+		status := groupErrorToHTTPStatus(err)
+		msg := "Failed to retrieve group"
+		if status == http.StatusNotFound {
+			msg = "Group not found"
+		}
+		writeJSONError(w, status, msg)
+		return
+	}
+	writeJSON(w, http.StatusOK, group)
+}
+
+func (h *GroupHandlers) DeleteGroup(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	if err := h.adminServer.DeleteGroup(r.Context(), name); err != nil {
+		glog.Errorf("Failed to delete group: %v", err)
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to delete group: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Group deleted successfully"})
+}
+
+func (h *GroupHandlers) GetGroupMembers(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	group, err := h.adminServer.GetGroupDetails(r.Context(), name)
+	if err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to get group: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"members": group.Members})
+}
+
+func (h *GroupHandlers) AddGroupMember(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	var req struct {
+		Username string `json:"username"`
+	}
+	if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
+		writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
+		return
+	}
+	if req.Username == "" {
+		writeJSONError(w, http.StatusBadRequest, "Username is required")
+		return
+	}
+	if err := h.adminServer.AddGroupMember(r.Context(), name, req.Username); err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to add member: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Member added successfully"})
+}
+
+func (h *GroupHandlers) RemoveGroupMember(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	username := mux.Vars(r)["username"]
+	if err := h.adminServer.RemoveGroupMember(r.Context(), name, username); err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to remove member: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Member removed successfully"})
+}
+
+func (h *GroupHandlers) GetGroupPolicies(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	group, err := h.adminServer.GetGroupDetails(r.Context(), name)
+	if err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to get group: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"policies": group.PolicyNames})
+}
+
+func (h *GroupHandlers) AttachGroupPolicy(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	var req struct {
+		PolicyName string `json:"policy_name"`
+	}
+	if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
+		writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
+		return
+	}
+	if req.PolicyName == "" {
+		writeJSONError(w, http.StatusBadRequest, "Policy name is required")
+		return
+	}
+	if err := h.adminServer.AttachGroupPolicy(r.Context(), name, req.PolicyName); err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to attach policy: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Policy attached successfully"})
+}
+
+func (h *GroupHandlers) DetachGroupPolicy(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	policyName := mux.Vars(r)["policyName"]
+	if err := h.adminServer.DetachGroupPolicy(r.Context(), name, policyName); err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to detach policy: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Policy detached successfully"})
+}
+
+func (h *GroupHandlers) SetGroupStatus(w http.ResponseWriter, r *http.Request) {
+	name := mux.Vars(r)["name"]
+	var req struct {
+		Enabled *bool `json:"enabled"`
+	}
+	if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
+		writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
+		return
+	}
+	if req.Enabled == nil {
+		writeJSONError(w, http.StatusBadRequest, "enabled field is required")
+		return
+	}
+	if err := h.adminServer.SetGroupStatus(r.Context(), name, *req.Enabled); err != nil {
+		writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to update group status: "+err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Group status updated"})
+}
+
+func (h *GroupHandlers) getGroupsPageData(r *http.Request) (dash.GroupsPageData, error) {
+	username := dash.UsernameFromContext(r.Context())
+	if username == "" {
+		username = "admin"
+	}
+
+	groups, err := h.adminServer.GetGroups(r.Context())
+	if err != nil {
+		return dash.GroupsPageData{}, err
+	}
+
+	activeCount := 0
+	for _, g := range groups {
+		if g.Status == "enabled" {
+			activeCount++
+		}
+	}
+
+	// Get available users for dropdown
+	var availableUsers []string
+	users, err := h.adminServer.GetObjectStoreUsers(r.Context())
+	if err == nil {
+		for _, user := range users {
+			availableUsers = append(availableUsers, user.Username)
+		}
+	}
+
+	// Get available policies for dropdown
+	var availablePolicies []string
+	policies, err := h.adminServer.GetPolicies()
+	if err == nil {
+		for _, p := range policies {
+			availablePolicies = append(availablePolicies, p.Name)
+		}
+	}
+
+	return dash.GroupsPageData{
+		Username:          username,
+		Groups:            groups,
+		TotalGroups:       len(groups),
+		ActiveGroups:      activeCount,
+		AvailableUsers:    availableUsers,
+		AvailablePolicies: availablePolicies,
+		LastUpdated:       time.Now(),
+	}, nil
+}
@@ -53,7 +53,8 @@ func (h *PluginHandlers) ShowPluginMonitoring(w http.ResponseWriter, r *http.Req
 }

 func (h *PluginHandlers) renderPluginPage(w http.ResponseWriter, r *http.Request, page string) {
-	component := app.Plugin(page)
+	initialJob := r.URL.Query().Get("job")
+	component := app.Plugin(page, initialJob)
 	viewCtx := layout.NewViewContext(r, dash.UsernameFromContext(r.Context()), dash.CSRFTokenFromContext(r.Context()))
 	layoutComponent := layout.Layout(viewCtx, component)

@@ -1,7 +1,9 @@
 package handlers

 import (
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"time"

@@ -155,10 +157,30 @@ func (h *UserHandlers) CreateAccessKey(w http.ResponseWriter, r *http.Request) {
 		return
 	}

-	accessKey, err := h.adminServer.CreateAccessKey(username)
+	var req *dash.CreateAccessKeyRequest
+	var body dash.CreateAccessKeyRequest
+	if err := decodeJSONBody(newJSONMaxReader(w, r), &body); err != nil {
+		if !errors.Is(err, io.EOF) {
+			writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
+			return
+		}
+		// Empty body: auto-generate both keys
+	} else {
+		req = &body
+	}
+
+	accessKey, err := h.adminServer.CreateAccessKey(username, req)
 	if err != nil {
 		glog.Errorf("Failed to create access key for user %s: %v", username, err)
-		writeJSONError(w, http.StatusInternalServerError, "Failed to create access key: "+err.Error())
+		if errors.Is(err, dash.ErrAccessKeyInUse) {
+			writeJSONError(w, http.StatusConflict, err.Error())
+		} else if errors.Is(err, dash.ErrUserNotFound) {
+			writeJSONError(w, http.StatusNotFound, err.Error())
+		} else if errors.Is(err, dash.ErrInvalidInput) {
+			writeJSONError(w, http.StatusBadRequest, err.Error())
+		} else {
+			writeJSONError(w, http.StatusInternalServerError, "Failed to create access key: "+err.Error())
+		}
 		return
 	}

@@ -297,7 +297,7 @@ func (mm *MaintenanceManager) logTopologyStatus() {
 	errorCount := mm.errorCount
 	mm.mutex.RUnlock()

-	glog.V(0).Infof("Topology status: %d nodes, %d disks, %d workers, %d pending tasks, %d running tasks, errors: %d",
+	glog.V(1).Infof("Topology status: %d nodes, %d disks, %d workers, %d pending tasks, %d running tasks, errors: %d",
 		nodeCount, diskCount, workerCount,
 		stats.TasksByStatus[TaskStatusPending],
 		stats.TasksByStatus[TaskStatusInProgress]+stats.TasksByStatus[TaskStatusAssigned],
@@ -30,7 +30,6 @@ const (
 	runsJSONFileName        = "runs.json"
 	trackedJobsJSONFileName = "tracked_jobs.json"
 	activitiesJSONFileName  = "activities.json"
-	schedulerJSONFileName   = "scheduler.json"
 	defaultDirPerm          = 0o755
 	defaultFilePerm         = 0o644
 )
@@ -54,7 +53,6 @@ type ConfigStore struct {
 	memTrackedJobs []TrackedJob
 	memActivities  []JobActivity
 	memJobDetails  map[string]TrackedJob
-	memScheduler   *SchedulerConfig
 }

 func NewConfigStore(adminDataDir string) (*ConfigStore, error) {
@@ -95,60 +93,6 @@ func (s *ConfigStore) BaseDir() string {
 	return s.baseDir
 }

-func (s *ConfigStore) LoadSchedulerConfig() (*SchedulerConfig, error) {
-	s.mu.RLock()
-	if !s.configured {
-		cfg := s.memScheduler
-		s.mu.RUnlock()
-		if cfg == nil {
-			return nil, nil
-		}
-		clone := *cfg
-		return &clone, nil
-	}
-	s.mu.RUnlock()
-
-	path := filepath.Join(s.baseDir, schedulerJSONFileName)
-	data, err := os.ReadFile(path)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return nil, nil
-		}
-		return nil, fmt.Errorf("read scheduler config: %w", err)
-	}
-
-	var cfg SchedulerConfig
-	if err := json.Unmarshal(data, &cfg); err != nil {
-		return nil, fmt.Errorf("unmarshal scheduler config: %w", err)
-	}
-	return &cfg, nil
-}
-
-func (s *ConfigStore) SaveSchedulerConfig(config *SchedulerConfig) error {
-	if config == nil {
-		return fmt.Errorf("scheduler config is nil")
-	}
-	normalized := normalizeSchedulerConfig(*config)
-
-	s.mu.Lock()
-	if !s.configured {
-		s.memScheduler = &normalized
-		s.mu.Unlock()
-		return nil
-	}
-	s.mu.Unlock()
-
-	payload, err := json.MarshalIndent(normalized, "", "  ")
-	if err != nil {
-		return fmt.Errorf("marshal scheduler config: %w", err)
-	}
-	path := filepath.Join(s.baseDir, schedulerJSONFileName)
-	if err := os.WriteFile(path, payload, defaultFilePerm); err != nil {
-		return fmt.Errorf("save scheduler config: %w", err)
-	}
-	return nil
-}
-
 func (s *ConfigStore) SaveDescriptor(jobType string, descriptor *plugin_pb.JobTypeDescriptor) error {
 	if descriptor == nil {
 		return fmt.Errorf("descriptor is nil")
@@ -78,8 +78,6 @@ type Plugin struct {
 	schedulerRun              map[string]*schedulerRunInfo
 	schedulerLoopMu           sync.Mutex
 	schedulerLoopState        schedulerLoopState
-	schedulerConfigMu         sync.RWMutex
-	schedulerConfig           SchedulerConfig
 	schedulerWakeCh           chan struct{}

 	dedupeMu           sync.Mutex
@@ -188,21 +186,6 @@ func New(options Options) (*Plugin, error) {
 	}
 	plugin.ctx, plugin.ctxCancel = context.WithCancel(context.Background())

-	if cfg, err := plugin.store.LoadSchedulerConfig(); err != nil {
-		glog.Warningf("Plugin failed to load scheduler config: %v", err)
-		plugin.schedulerConfig = DefaultSchedulerConfig()
-	} else if cfg == nil {
-		defaults := DefaultSchedulerConfig()
-		plugin.schedulerConfig = defaults
-		if plugin.store.IsConfigured() {
-			if err := plugin.store.SaveSchedulerConfig(&defaults); err != nil {
-				glog.Warningf("Plugin failed to persist scheduler defaults: %v", err)
-			}
-		}
-	} else {
-		plugin.schedulerConfig = normalizeSchedulerConfig(*cfg)
-	}
-
 	if err := plugin.loadPersistedMonitorState(); err != nil {
 		glog.Warningf("Plugin failed to load persisted monitoring state: %v", err)
 	}
@@ -409,7 +392,6 @@ func (r *Plugin) SaveJobTypeConfig(config *plugin_pb.PersistedJobTypeConfig) err
 	return nil
 }

-
 func (r *Plugin) LoadDescriptor(jobType string) (*plugin_pb.JobTypeDescriptor, error) {
 	return r.store.LoadDescriptor(jobType)
 }
@@ -426,31 +408,6 @@ func (r *Plugin) BaseDir() string {
 	return r.store.BaseDir()
 }

-func (r *Plugin) GetSchedulerConfig() SchedulerConfig {
-	if r == nil {
-		return DefaultSchedulerConfig()
-	}
-	r.schedulerConfigMu.RLock()
-	cfg := r.schedulerConfig
-	r.schedulerConfigMu.RUnlock()
-	return normalizeSchedulerConfig(cfg)
-}
-
-func (r *Plugin) UpdateSchedulerConfig(cfg SchedulerConfig) (SchedulerConfig, error) {
-	if r == nil {
-		return DefaultSchedulerConfig(), fmt.Errorf("plugin is not initialized")
-	}
-	normalized := normalizeSchedulerConfig(cfg)
-	if err := r.store.SaveSchedulerConfig(&normalized); err != nil {
-		return SchedulerConfig{}, err
-	}
-	r.schedulerConfigMu.Lock()
-	r.schedulerConfig = normalized
-	r.schedulerConfigMu.Unlock()
-	r.wakeScheduler()
-	return normalized, nil
-}
-
 func (r *Plugin) acquireAdminLock(reason string) (func(), error) {
 	if r == nil || r.lockManager == nil {
 		return func() {}, nil
@@ -730,6 +687,11 @@ func (r *Plugin) executeJobWithExecutor(
 	}
 }

+// HasCapableWorker checks if any non-stale worker has a capability for the given job type.
+func (r *Plugin) HasCapableWorker(jobType string) bool {
+	return r.registry.HasCapableWorker(jobType)
+}
+
 func (r *Plugin) ListWorkers() []*WorkerSession {
 	return r.registry.List()
 }
@@ -1456,3 +1418,13 @@ func (s *streamSession) close() {
 		close(s.outgoing)
 	})
 }
+
+// WorkerConnectForTest simulates a worker connecting (test helper).
+func (r *Plugin) WorkerConnectForTest(hello *plugin_pb.WorkerHello) {
+	r.registry.UpsertFromHello(hello)
+}
+
+// WorkerDisconnectForTest simulates a worker disconnecting (test helper).
+func (r *Plugin) WorkerDisconnectForTest(workerID string) {
+	r.registry.Remove(workerID)
+}
@@ -32,6 +32,7 @@ const (
 	defaultClusterContextTimeout               = 10 * time.Second
 	defaultWaitingBacklogFloor                 = 8
 	defaultWaitingBacklogMultiplier            = 4
+	maxEstimatedRuntimeCap                     = 8 * time.Hour
 )

 type schedulerPolicy struct {
@@ -64,7 +65,7 @@ func (r *Plugin) schedulerLoop() {
 		}

 		r.setSchedulerLoopState("", "sleeping")
-		idleSleep := r.GetSchedulerConfig().IdleSleepDuration()
+		idleSleep := defaultSchedulerIdleSleep
 		if nextRun := r.earliestNextDetectionAt(); !nextRun.IsZero() {
 			if until := time.Until(nextRun); until <= 0 {
 				idleSleep = 0
@@ -112,7 +113,6 @@ func (r *Plugin) runSchedulerIteration() bool {
 	}

 	active := make(map[string]struct{}, len(jobTypes))
-	schedulerIdleSleep := r.GetSchedulerConfig().IdleSleepDuration()
 	hadJobs := false

 	for _, jobType := range jobTypes {
@@ -129,7 +129,7 @@ func (r *Plugin) runSchedulerIteration() bool {
 		}
 		initialDelay := time.Duration(0)
 		if runInfo := r.snapshotSchedulerRun(jobType); runInfo.lastRunStartedAt.IsZero() {
-			initialDelay = schedulerIdleSleep / 2
+			initialDelay = 5 * time.Second
 		}
 		if !r.markDetectionDue(jobType, policy.DetectionInterval, initialDelay) {
 			continue
@@ -294,6 +294,26 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo

 	r.setSchedulerLoopState(jobType, "executing")

+	// Scan proposals for the maximum estimated_runtime_seconds so the
+	// execution phase gets enough time for large jobs (e.g. vacuum on
+	// big volumes). If any proposal needs more time than the remaining
+	// JobTypeMaxRuntime, extend the execution context accordingly.
+	var maxEstimatedRuntime time.Duration
+	for _, p := range filtered {
+		if p.Parameters != nil {
+			if est, ok := p.Parameters["estimated_runtime_seconds"]; ok {
+				if v := est.GetInt64Value(); v > 0 {
+					if d := time.Duration(v) * time.Second; d > maxEstimatedRuntime {
+						maxEstimatedRuntime = d
+					}
+				}
+			}
+		}
+	}
+	if maxEstimatedRuntime > maxEstimatedRuntimeCap {
+		maxEstimatedRuntime = maxEstimatedRuntimeCap
+	}
+
 	remaining = time.Until(start.Add(maxRuntime))
 	if remaining <= 0 {
 		r.appendActivity(JobActivity{
@@ -307,6 +327,17 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 		return detected
 	}

+	// If the longest estimated job exceeds the remaining JobTypeMaxRuntime,
+	// create a new execution context with enough headroom instead of using
+	// jobCtx which would cancel too early.
+	execCtx := jobCtx
+	execCancel := context.CancelFunc(func() {})
+	if maxEstimatedRuntime > 0 && maxEstimatedRuntime > remaining {
+		execCtx, execCancel = context.WithTimeout(context.Background(), maxEstimatedRuntime)
+		remaining = maxEstimatedRuntime
+	}
+	defer execCancel()
+
 	execPolicy := policy
 	if execPolicy.ExecutionTimeout <= 0 {
 		execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout
@@ -315,10 +346,10 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 		execPolicy.ExecutionTimeout = remaining
 	}

-	successCount, errorCount, canceledCount := r.dispatchScheduledProposals(jobCtx, jobType, filtered, clusterContext, execPolicy)
+	successCount, errorCount, canceledCount := r.dispatchScheduledProposals(execCtx, jobType, filtered, clusterContext, execPolicy)

 	status := "success"
-	if jobCtx.Err() != nil {
+	if execCtx.Err() != nil {
 		status = "timeout"
 	} else if errorCount > 0 || canceledCount > 0 {
 		status = "error"
@@ -938,7 +969,24 @@ func (r *Plugin) executeScheduledJobWithExecutor(
 		if parent == nil {
 			parent = context.Background()
 		}
-		execCtx, cancel := context.WithTimeout(parent, policy.ExecutionTimeout)
+		// Use the job's estimated runtime if provided and larger than the
+		// default execution timeout. This lets handlers like vacuum scale
+		// the timeout based on volume size so large volumes are not killed.
+		timeout := policy.ExecutionTimeout
+		if job.Parameters != nil {
+			if est, ok := job.Parameters["estimated_runtime_seconds"]; ok {
+				if v := est.GetInt64Value(); v > 0 {
+					estimated := time.Duration(v) * time.Second
+					if estimated > maxEstimatedRuntimeCap {
+						estimated = maxEstimatedRuntimeCap
+					}
+					if estimated > timeout {
+						timeout = estimated
+					}
+				}
+			}
+		}
+		execCtx, cancel := context.WithTimeout(parent, timeout)
 		_, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt))
 		cancel()
 		if err == nil {
@@ -1135,22 +1183,6 @@ func secondsFromDuration(duration time.Duration) int32 {
 	return int32(duration / time.Second)
 }

-func waitForShutdownOrTimer(shutdown <-chan struct{}, duration time.Duration) bool {
-	if duration <= 0 {
-		return true
-	}
-
-	timer := time.NewTimer(duration)
-	defer timer.Stop()
-
-	select {
-	case <-shutdown:
-		return false
-	case <-timer.C:
-		return true
-	}
-}
-
 func waitForShutdownOrTimerWithContext(shutdown <-chan struct{}, ctx context.Context, duration time.Duration) bool {
 	if duration <= 0 {
 		return true
--- a/Show More
+++ b/Show More