guard source deletion on ec success

* Prevent concurrent maintenance tasks per volume

* fix panic

* fix(s3api): correctly extract host header port when X-Forwarded-Port is present

* test(s3api): add test cases for misreported X-Forwarded-Port

.github/workflows/s3-proxy-signature-tests.yml

@@ -0,0 +1,122 @@
+name: "S3 Proxy Signature Tests"
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+concurrency:
+  group: ${{ github.head_ref || github.ref }}/s3-proxy-signature-tests
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  proxy-signature-tests:
+    name: S3 Proxy Signature Verification Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 15
+    steps:
+      - name: Check out code into the Go module directory
+        uses: actions/checkout@v6
+
+      - name: Set up Go 1.x
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+        id: go
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build SeaweedFS binary for Linux
+        run: |
+          set -x
+          CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -buildvcs=false -v -o test/s3/proxy_signature/weed ./weed
+
+      - name: Run S3 Proxy Signature Tests
+        timeout-minutes: 10
+        working-directory: test/s3/proxy_signature
+        run: |
+          set -x
+          echo "Starting Docker Compose services..."
+          docker compose up -d --build
+          
+          # Check if containers are running
+          echo "Checking container status..."
+          docker compose ps
+          
+          # Wait for services to be ready
+          echo "Waiting for nginx proxy to be ready..."
+          PROXY_READY=0
+          for i in $(seq 1 30); do
+            if curl -s http://localhost:9000/ > /dev/null 2>&1; then
+              echo "Proxy is ready"
+              PROXY_READY=1
+              break
+            fi
+            echo "Waiting for proxy... ($i/30)"
+            sleep 1
+          done
+          if [ $PROXY_READY -eq 0 ]; then
+            echo "ERROR: Proxy failed to become ready after 30 seconds"
+            echo "Docker compose logs:"
+            docker compose logs --no-color || true
+            exit 1
+          fi
+          
+          # Wait for SeaweedFS to be ready
+          echo "Waiting for SeaweedFS S3 gateway to be ready via proxy..."
+          S3_READY=0
+          for i in $(seq 1 30); do
+            # Check logs first for startup message (weed mini says "S3 service is ready")
+            if docker compose logs seaweedfs 2>&1 | grep -qE "S3 (gateway|service).*(started|ready)"; then
+              echo "SeaweedFS S3 gateway is ready"
+              S3_READY=1
+              break
+            fi
+            # Fallback: check headers via proxy (which is already ready)
+            if curl -s -I http://localhost:9000/ | grep -qi "SeaweedFS"; then
+              echo "SeaweedFS S3 gateway is responding via proxy"
+              S3_READY=1
+              break
+            fi
+            echo "Waiting for S3 gateway... ($i/30)"
+            sleep 1
+          done
+          if [ $S3_READY -eq 0 ]; then
+            echo "ERROR: SeaweedFS S3 gateway failed to become ready after 30 seconds"
+            echo "Latest seaweedfs logs:"
+            docker compose logs --no-color --tail 20 seaweedfs || true
+            exit 1
+          fi
+          
+          # Run the test script inside AWS CLI container
+          echo "Running test script..."
+          docker run --rm --network host \
+            --entrypoint bash \
+            amazon/aws-cli:latest \
+            -c "$(cat test.sh)"
+          
+          TEST_RESULT=$?
+          
+          # Cleanup
+          docker compose down
+          
+          exit $TEST_RESULT
+
+      - name: Cleanup on failure
+        if: failure()
+        working-directory: test/s3/proxy_signature
+        run: |
+          echo "Cleaning up Docker containers..."
+          ls -al weed || true
+          ldd weed || true
+          echo "Docker compose logs:"
+          docker compose logs --no-color || true
+          echo "Container status before cleanup:"
+          docker ps -a
+          echo "Stopping services..."
+          docker compose down || true

k8s/charts/seaweedfs/templates/cosi/cosi-bucket-class.yaml

@@ -6,6 +6,10 @@ metadata:
   name: {{ .Values.cosi.bucketClassName }}
 driverName: {{ .Values.cosi.driverName }}
 deletionPolicy: Delete
+{{- with .Values.cosi.bucketClassParameters }}
+parameters:
+  {{- toYaml . | nindent 2 }}
+{{- end }}
 ---
 kind: BucketAccessClass
 apiVersion: objectstorage.k8s.io/v1alpha1

k8s/charts/seaweedfs/templates/shared/post-install-bucket-hook.yaml

@@ -3,6 +3,15 @@
 {{- $s3Enabled := false }}
 {{- $enableAuth := false }}
 {{- $existingConfigSecret := "" }}
+{{- $bucketsFolder := "/buckets" }}
+{{- $bucketEnvVars := merge (dict) (.Values.global.extraEnvironmentVars | default dict) }}
+{{- if .Values.allInOne.enabled }}
+  {{- $bucketEnvVars = merge (.Values.allInOne.extraEnvironmentVars | default dict) $bucketEnvVars }}
+{{- else }}
+  {{- $bucketEnvVars = merge (.Values.filer.extraEnvironmentVars | default dict) $bucketEnvVars }}
+{{- end }}
+{{- $bucketsFolder = default $bucketsFolder (get $bucketEnvVars "WEED_FILER_BUCKETS_FOLDER") }}
+{{- $bucketsFolder = trimSuffix "/" $bucketsFolder }}
 
 {{- /* Check allInOne mode first */}}
 {{- if .Values.allInOne.enabled }}
@@ -108,15 +117,52 @@ spec:
             wait_for_service "http://$WEED_CLUSTER_SW_MASTER{{ .Values.master.readinessProbe.httpGet.path }}"
             wait_for_service "http://$WEED_CLUSTER_SW_FILER{{ .Values.filer.readinessProbe.httpGet.path }}"
             {{- end }}
+          set -o pipefail
           {{- range $createBuckets }}
-            if /bin/echo "s3.bucket.list" | /usr/bin/weed shell | awk '{print $1}' | grep -Fxq "{{ .name }}"; then
-              echo "Bucket '{{ .name }}' already exists, skipping creation."
+            {{- $bucketName := .name }}
+            {{- $bucketLock := or .lock .objectLock .withLock }}
+            bucket_list=$(/bin/echo 's3.bucket.list' | /usr/bin/weed shell) || { echo "Error listing s3 buckets"; exit 1; }
+            if echo "$bucket_list" | awk '{print $1}' | grep -Fxq "{{ $bucketName }}"; then
+              echo "Bucket '{{ $bucketName }}' already exists, skipping creation."
             else
-              echo "Creating bucket '{{ .name }}'..."
-              /bin/echo "s3.bucket.create --name {{ .name }}" | /usr/bin/weed shell
+              echo "Creating bucket '{{ $bucketName }}'..."
+              /bin/echo 's3.bucket.create --name {{ $bucketName }}{{- if $bucketLock }} --withLock{{- end }}' | /usr/bin/weed shell
             fi
           {{- end }}
           {{- range $createBuckets }}
+          {{- $bucketLock := or .lock .objectLock .withLock }}
+          {{- if $bucketLock }}
+            /bin/echo 's3.bucket.lock -name {{ .name }} -enable' | /usr/bin/weed shell
+          {{- end }}
+          {{- end }}
+          {{- range $createBuckets }}
+          {{- $bucketVersioning := "" }}
+          {{- if kindIs "bool" .versioning }}
+            {{- if .versioning }}
+              {{- $bucketVersioning = "Enabled" }}
+            {{- end }}
+          {{- else if kindIs "string" .versioning }}
+            {{- $versioningLower := lower .versioning }}
+            {{- if eq $versioningLower "enabled" "enable" "true" }}
+              {{- $bucketVersioning = "Enabled" }}
+            {{- else if eq $versioningLower "suspended" "disable" "false" }}
+              {{- $bucketVersioning = "Suspended" }}
+            {{- else if or (eq .versioning "Enabled") (eq .versioning "Suspended") }}
+              {{- $bucketVersioning = .versioning }}
+            {{- else }}
+              {{- fail (printf "Invalid versioning value for bucket %s: %s. Must be 'Enabled' or 'Suspended'" .name .versioning) }}
+            {{- end }}
+          {{- end }}
+          {{- if $bucketVersioning }}
+            /bin/echo 's3.bucket.versioning -name {{ .name }} -status {{ $bucketVersioning }}' | /usr/bin/weed shell
+          {{- end }}
+          {{- end }}
+          {{- range $createBuckets }}
+          {{- if .ttl }}
+            /bin/echo 'fs.configure -locationPrefix={{ $bucketsFolder }}/{{ .name }}/ -ttl={{ .ttl }} -apply' | /usr/bin/weed shell
+          {{- end }}
+          {{- end }}
+          {{- range $createBuckets }}
           {{- if .anonymousRead }}
             /bin/echo \
             "s3.configure --user anonymous \

k8s/charts/seaweedfs/values.yaml

@@ -898,9 +898,15 @@ filer:
     auditLogConfig: {}
     # You may specify buckets to be created during the install or upgrade process.
     # Buckets may be exposed publicly by setting `anonymousRead` to `true`
+    # ttl format: [1-255][m|h|d|w|M|y] (e.g., 7d)
+    # objectLock enables S3 Object Lock (irreversible, forces versioning)
+    # versioning: Enabled or Suspended (or true to enable)
     # createBuckets:
     #   - name: bucket-a
     #     anonymousRead: true
+    #     ttl: 7d
+    #     objectLock: true
+    #     versioning: Enabled
     #   - name: bucket-b
     #     anonymousRead: false
 
@@ -936,9 +942,15 @@ s3:
   auditLogConfig: {}
   # You may specify buckets to be created during the install or upgrade process.
   # Buckets may be exposed publicly by setting `anonymousRead` to `true`
+  # ttl format: [1-255][m|h|d|w|M|y] (e.g., 7d)
+  # objectLock enables S3 Object Lock (irreversible, forces versioning)
+  # versioning: Enabled or Suspended (or true to enable)
   # createBuckets:
   #   - name: bucket-a
   #     anonymousRead: true
+  #     ttl: 7d
+  #     objectLock: true
+  #     versioning: Enabled
   #   - name: bucket-b
   #     anonymousRead: false
 
@@ -1443,9 +1455,15 @@ allInOne:
     auditLogConfig: null  # S3 audit log configuration (null inherits from s3.auditLogConfig)
     # 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)
+    # objectLock enables S3 Object Lock (irreversible, forces versioning)
+    # versioning: Enabled or Suspended (or true to enable)
     # createBuckets:
     #   - name: bucket-a
     #     anonymousRead: true
+    #     ttl: 7d
+    #     objectLock: true
+    #     versioning: Enabled
     #   - name: bucket-b
     #     anonymousRead: false
 
@@ -1611,6 +1629,8 @@ cosi:
   image: "ghcr.io/seaweedfs/seaweedfs-cosi-driver:v0.1.2"
   driverName: "seaweedfs.objectstorage.k8s.io"
   bucketClassName: "seaweedfs"
+  # Optional parameters to pass to the default BucketClass (e.g., diskType for tiered storage)
+  bucketClassParameters: {}
   endpoint: ""
   region: ""
 

test/s3/proxy_signature/Dockerfile

@@ -0,0 +1,12 @@
+FROM alpine:3.20
+RUN apk add --no-cache curl && \
+    addgroup -S seaweed && \
+    adduser -S seaweed -G seaweed
+COPY weed /usr/bin/weed
+RUN chmod +x /usr/bin/weed && \
+    chown seaweed:seaweed /usr/bin/weed && \
+    mkdir -p /etc/seaweedfs /data/filerldb2 && \
+    chown -R seaweed:seaweed /etc/seaweedfs /data && \
+    chmod 755 /data /etc/seaweedfs /data/filerldb2
+WORKDIR /data
+USER seaweed

test/s3/proxy_signature/README.md

@@ -0,0 +1,79 @@
+# S3 Proxy Signature Verification Test
+
+Integration test that verifies S3 signature verification works correctly when
+SeaweedFS is deployed behind a reverse proxy (nginx).
+
+## What it tests
+
+- S3 operations (create bucket, put/get/head/list/delete) through an nginx
+  reverse proxy with `X-Forwarded-Host`, `X-Forwarded-Port`, and
+  `X-Forwarded-Proto` headers
+- SeaweedFS configured with `-s3.externalUrl=http://localhost:9000` so the
+  signature verification uses the client-facing host instead of the internal
+  backend address
+
+## Architecture
+
+```text
+AWS CLI (signs with Host: localhost:9000)
+    |
+    v
+nginx (:9000)
+    |  proxy_pass → seaweedfs:8333
+    |  Sets: X-Forwarded-Host: localhost
+    |         X-Forwarded-Port: 9000
+    |         X-Forwarded-Proto: http
+    v
+SeaweedFS S3 (:8333, -s3.externalUrl=http://localhost:9000)
+    |  externalHost = "localhost:9000" (parsed at startup)
+    |  extractHostHeader() returns "localhost:9000"
+    |  Matches what AWS CLI signed with
+    v
+Signature verification succeeds
+```
+
+**Note:** When `-s3.externalUrl` is configured, direct access to the backend
+port (8333) will fail signature verification because the client signs with a
+different Host header than what `externalUrl` specifies. This is expected —
+all S3 traffic should go through the proxy.
+
+## Prerequisites
+
+- Docker and Docker Compose
+- AWS CLI v2 (on host or via Docker, see below)
+
+## Running
+
+```bash
+# Build the weed binary first (from repo root):
+cd /path/to/seaweedfs
+go build -o test/s3/proxy_signature/weed ./weed
+cd test/s3/proxy_signature
+
+# Start services
+docker compose up -d --build
+
+# Option A: Run test with aws CLI installed locally
+./test.sh
+
+# Option B: Run test without aws CLI (uses Docker container)
+docker run --rm --network host --entrypoint "" amazon/aws-cli:latest \
+    bash < test.sh
+
+# Tear down
+docker compose down
+
+# Clean up the weed binary
+rm -f weed
+```
+
+## Troubleshooting
+
+If signature verification fails through the proxy, check:
+
+1. nginx is setting `X-Forwarded-Host` and `X-Forwarded-Port` correctly
+2. SeaweedFS is started with `-s3.externalUrl` matching the client endpoint
+3. The AWS CLI endpoint URL matches the proxy address
+
+You can also set the `S3_EXTERNAL_URL` environment variable instead of the
+`-s3.externalUrl` flag.

test/s3/proxy_signature/docker-compose.yml

@@ -0,0 +1,28 @@
+services:
+  seaweedfs:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: >
+      /usr/bin/weed mini
+      -s3.config=/etc/seaweedfs/s3.json
+      -s3.externalUrl=http://localhost:9000
+      -ip=seaweedfs
+    volumes:
+      - ./s3.json:/etc/seaweedfs/s3.json:ro
+    healthcheck:
+      test: ["CMD", "curl", "-sf", "http://seaweedfs:8333/status"]
+      interval: 3s
+      timeout: 2s
+      retries: 20
+      start_period: 5s
+
+  nginx:
+    image: nginx:alpine
+    ports:
+      - "9000:9000"
+    volumes:
+      - ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
+    depends_on:
+      seaweedfs:
+        condition: service_healthy

test/s3/proxy_signature/nginx.conf

@@ -0,0 +1,23 @@
+server {
+    listen 9000;
+    server_name localhost;
+
+    # Allow large uploads
+    client_max_body_size 64m;
+
+    location / {
+        proxy_pass http://seaweedfs:8333;
+
+        # Standard reverse proxy headers — this is what Kong, Traefik, etc. do
+        proxy_set_header Host              $host:$server_port;
+        proxy_set_header X-Real-IP         $remote_addr;
+        proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Host  $host;
+        proxy_set_header X-Forwarded-Port  $server_port;
+        proxy_set_header X-Forwarded-Proto $scheme;
+
+        proxy_http_version 1.1;
+        proxy_buffering off;
+        proxy_request_buffering off;
+    }
+}

test/s3/proxy_signature/s3.json

@@ -0,0 +1,20 @@
+{
+  "identities": [
+    {
+      "name": "test_admin",
+      "credentials": [
+        {
+          "accessKey": "test_access_key",
+          "secretKey": "test_secret_key"
+        }
+      ],
+      "actions": [
+        "Admin",
+        "Read",
+        "List",
+        "Tagging",
+        "Write"
+      ]
+    }
+  ]
+}

test/s3/proxy_signature/test.sh

@@ -0,0 +1,132 @@
+#!/usr/bin/env bash
+#
+# Integration test for S3 signature verification behind a reverse proxy.
+#
+# Usage:
+#   # With aws CLI installed locally:
+#   docker compose up -d --build && ./test.sh && docker compose down
+#
+#   # Without aws CLI (runs test inside a container):
+#   docker compose up -d --build
+#   docker run --rm --network host --entrypoint "" amazon/aws-cli:latest \
+#       bash < test.sh
+#   docker compose down
+#
+# This script tests S3 operations through an nginx reverse proxy to verify
+# that signature verification works correctly when SeaweedFS is configured
+# with -s3.externalUrl=http://localhost:9000.
+#
+set -euo pipefail
+
+PROXY_ENDPOINT="http://localhost:9000"
+ACCESS_KEY="test_access_key"
+SECRET_KEY="test_secret_key"
+REGION="us-east-1"
+BUCKET="test-proxy-sig-$$"
+
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+NC='\033[0m'
+
+pass() { echo -e "${GREEN}PASS${NC}: $1"; }
+fail() { echo -e "${RED}FAIL${NC}: $1"; exit 1; }
+
+# Helper: run aws s3api command against a given endpoint
+s3() {
+    local endpoint="$1"
+    shift
+    aws s3api \
+        --endpoint-url "$endpoint" \
+        --region "$REGION" \
+        --no-verify-ssl \
+        "$@" 2>&1
+}
+
+export AWS_ACCESS_KEY_ID="$ACCESS_KEY"
+export AWS_SECRET_ACCESS_KEY="$SECRET_KEY"
+
+echo "=== S3 Proxy Signature Verification Test ==="
+echo ""
+echo "Testing S3 access through nginx reverse proxy at $PROXY_ENDPOINT"
+echo "SeaweedFS configured with -s3.externalUrl=http://localhost:9000"
+echo "AWS CLI signs requests with Host: localhost:9000"
+echo ""
+
+# Wait for proxy to be ready
+echo "Waiting for nginx proxy to be ready..."
+for i in $(seq 1 30); do
+    # Use aws CLI for health check if curl is missing
+    if command -v curl >/dev/null 2>&1; then
+        http_code=$(curl -s -o /dev/null -w "%{http_code}" "$PROXY_ENDPOINT/" 2>/dev/null || echo "000")
+        case $http_code in
+            200|403|405) break ;;
+        esac
+    else
+        if aws s3api list-buckets --endpoint-url "$PROXY_ENDPOINT" --no-sign-request >/dev/null 2>&1; then
+            break
+        fi
+    fi
+    if [ "$i" -eq 30 ]; then
+        fail "Proxy did not become ready in time"
+    fi
+    echo "Waiting for proxy $i/30..."
+    sleep 1
+done
+echo "Proxy is ready."
+echo ""
+
+# --- Test 1: Bucket operations through proxy ---
+echo "--- Test 1: Bucket operations through proxy ---"
+s3 "$PROXY_ENDPOINT" create-bucket --bucket "$BUCKET" > /dev/null \
+    && pass "create-bucket" \
+    || fail "create-bucket — signature verification likely failed"
+
+s3 "$PROXY_ENDPOINT" list-buckets > /dev/null \
+    && pass "list-buckets" \
+    || fail "list-buckets"
+echo ""
+
+# --- Test 2: Object CRUD through proxy ---
+echo "--- Test 2: Object CRUD through proxy ---"
+echo "hello-from-proxy" > /tmp/test-proxy-sig.txt
+
+s3 "$PROXY_ENDPOINT" put-object --bucket "$BUCKET" --key "test.txt" --body /tmp/test-proxy-sig.txt > /dev/null \
+    && pass "put-object" \
+    || fail "put-object"
+
+s3 "$PROXY_ENDPOINT" head-object --bucket "$BUCKET" --key "test.txt" > /dev/null \
+    && pass "head-object" \
+    || fail "head-object"
+
+s3 "$PROXY_ENDPOINT" list-objects-v2 --bucket "$BUCKET" > /dev/null \
+    && pass "list-objects-v2" \
+    || fail "list-objects-v2"
+
+s3 "$PROXY_ENDPOINT" get-object --bucket "$BUCKET" --key "test.txt" /tmp/test-proxy-sig-get.txt > /dev/null \
+    && pass "get-object" \
+    || fail "get-object"
+
+# Verify content round-trip
+CONTENT=$(cat /tmp/test-proxy-sig-get.txt)
+if [ "$CONTENT" = "hello-from-proxy" ]; then
+    pass "content integrity (round-trip)"
+else
+    fail "content mismatch: got \"$CONTENT\", expected \"hello-from-proxy\""
+fi
+echo ""
+
+# --- Test 3: Delete operations through proxy ---
+echo "--- Test 3: Delete through proxy ---"
+s3 "$PROXY_ENDPOINT" delete-object --bucket "$BUCKET" --key "test.txt" > /dev/null \
+    && pass "delete-object" \
+    || fail "delete-object"
+
+s3 "$PROXY_ENDPOINT" delete-bucket --bucket "$BUCKET" > /dev/null \
+    && pass "delete-bucket" \
+    || fail "delete-bucket"
+echo ""
+
+# Cleanup temp files
+rm -f /tmp/test-proxy-sig.txt /tmp/test-proxy-sig-get.txt
+
+echo "=== All tests passed ==="

weed/admin/dash/admin_server.go

@@ -3,6 +3,11 @@ package dash
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/admin/maintenance"
 	adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin"
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
@@ -19,10 +24,6 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"google.golang.org/grpc"
-	"net/http"
-	"sort"
-	"strings"
-	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/s3api"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
@@ -980,7 +981,7 @@ func (s *AdminServer) GetPluginRunHistory(jobType string) (*adminplugin.JobTypeR
 }
 
 // ListPluginJobTypes returns known plugin job types from connected worker registry and persisted data.
-func (s *AdminServer) ListPluginJobTypes() ([]string, error) {
+func (s *AdminServer) ListPluginJobTypes() ([]adminplugin.JobTypeInfo, error) {
 	if s.plugin == nil {
 		return nil, fmt.Errorf("plugin is not enabled")
 	}

weed/admin/maintenance/maintenance_queue.go

@@ -131,6 +131,14 @@ func (mq *MaintenanceQueue) cleanupCompletedTasks() {
 func (mq *MaintenanceQueue) AddTask(task *MaintenanceTask) {
 	mq.mutex.Lock()
 
+	// Enforce one queued/active task per volume (across all task types).
+	if mq.hasQueuedOrActiveTaskForVolume(task.VolumeID) {
+		mq.mutex.Unlock()
+		glog.V(1).Infof("Task skipped (volume busy): %s for volume %d on %s (already queued or running)",
+			task.Type, task.VolumeID, task.Server)
+		return
+	}
+
 	// Check for duplicate tasks (same type + volume + not completed)
 	if mq.hasDuplicateTask(task) {
 		mq.mutex.Unlock()
@@ -188,6 +196,25 @@ func (mq *MaintenanceQueue) AddTask(task *MaintenanceTask) {
 		taskSnapshot.ID, taskSnapshot.Type, taskSnapshot.VolumeID, taskSnapshot.Server, taskSnapshot.Priority, scheduleInfo, taskSnapshot.Reason)
 }
 
+// hasQueuedOrActiveTaskForVolume checks if any pending/assigned/in-progress task already exists for this volume.
+// Caller must hold mq.mutex.
+func (mq *MaintenanceQueue) hasQueuedOrActiveTaskForVolume(volumeID uint32) bool {
+	if volumeID == 0 {
+		return false
+	}
+	for _, existingTask := range mq.tasks {
+		if existingTask.VolumeID != volumeID {
+			continue
+		}
+		if existingTask.Status == TaskStatusPending ||
+			existingTask.Status == TaskStatusAssigned ||
+			existingTask.Status == TaskStatusInProgress {
+			return true
+		}
+	}
+	return false
+}
+
 // hasDuplicateTask checks if a similar task already exists (same type, volume, and not completed)
 func (mq *MaintenanceQueue) hasDuplicateTask(newTask *MaintenanceTask) bool {
 	for _, existingTask := range mq.tasks {
@@ -260,6 +287,13 @@ func (mq *MaintenanceQueue) GetNextTask(workerID string, capabilities []Maintena
 			continue
 		}
 
+		// Avoid scheduling concurrent operations on the same volume
+		if activeTaskID, activeTaskType, hasActive := mq.activeTaskForVolume(task.VolumeID, task.ID); hasActive {
+			glog.V(2).Infof("Task %s (%s) skipped for worker %s: volume %d is busy with task %s (%s)",
+				task.ID, task.Type, workerID, task.VolumeID, activeTaskID, activeTaskType)
+			continue
+		}
+
 		// Check if worker can handle this task type
 		if !mq.workerCanHandle(task.Type, capabilities) {
 			glog.V(3).Infof("Task %s (%s) skipped for worker %s: capability mismatch (worker has: %v)", task.ID, task.Type, workerID, capabilities)
@@ -304,6 +338,14 @@ func (mq *MaintenanceQueue) GetNextTask(workerID string, capabilities []Maintena
 		return nil
 	}
 
+	// Re-check volume conflict after acquiring write lock
+	if activeTaskID, activeTaskType, hasActive := mq.activeTaskForVolume(selectedTask.VolumeID, selectedTaskID); hasActive {
+		mq.mutex.Unlock()
+		glog.V(2).Infof("Task %s no longer available for worker %s: volume %d is busy with task %s (%s)",
+			selectedTaskID, workerID, selectedTask.VolumeID, activeTaskID, activeTaskType)
+		return nil
+	}
+
 	// Record assignment history
 	workerAddress := ""
 	if worker, exists := mq.workers[workerID]; exists {
@@ -877,6 +919,28 @@ func (mq *MaintenanceQueue) workerCanHandle(taskType MaintenanceTaskType, capabi
 	return false
 }
 
+// activeTaskForVolume returns the active task ID/type for a volume, if any.
+// Caller must hold mq.mutex (read or write).
+func (mq *MaintenanceQueue) activeTaskForVolume(volumeID uint32, excludeTaskID string) (string, MaintenanceTaskType, bool) {
+	if volumeID == 0 {
+		return "", "", false
+	}
+
+	for _, task := range mq.tasks {
+		if task.ID == excludeTaskID {
+			continue
+		}
+		if task.VolumeID != volumeID {
+			continue
+		}
+		if task.Status == TaskStatusAssigned || task.Status == TaskStatusInProgress {
+			return task.ID, task.Type, true
+		}
+	}
+
+	return "", "", false
+}
+
 // canScheduleTaskNow determines if a task can be scheduled using task schedulers or fallback logic
 func (mq *MaintenanceQueue) canScheduleTaskNow(task *MaintenanceTask) bool {
 	glog.V(2).Infof("Checking if task %s (type: %s) can be scheduled", task.ID, task.Type)

weed/admin/maintenance/maintenance_queue_test.go

@@ -969,3 +969,154 @@ func TestMaintenanceQueue_AssignTaskRollback(t *testing.T) {
 		t.Errorf("Task %s should still be in pendingTasks slice", taskID)
 	}
 }
+
+func TestGetNextTask_SkipsVolumeConflictsAcrossTypes(t *testing.T) {
+	policy := &MaintenancePolicy{
+		TaskPolicies: map[string]*worker_pb.TaskPolicy{
+			"balance":        {MaxConcurrent: 2},
+			"erasure_coding": {MaxConcurrent: 2},
+			"vacuum":         {MaxConcurrent: 2},
+		},
+	}
+
+	mq := NewMaintenanceQueue(policy)
+
+	now := time.Now()
+	mq.AddTask(&MaintenanceTask{
+		ID:          "t1",
+		Type:        MaintenanceTaskType("balance"),
+		Priority:    PriorityHigh,
+		VolumeID:    100,
+		Server:      "server1",
+		ScheduledAt: now.Add(-3 * time.Second),
+	})
+	t2 := &MaintenanceTask{
+		ID:          "t2",
+		Type:        MaintenanceTaskType("erasure_coding"),
+		Priority:    PriorityNormal,
+		VolumeID:    100,
+		Server:      "server1",
+		Status:      TaskStatusPending,
+		ScheduledAt: now.Add(-2 * time.Second),
+	}
+	mq.mutex.Lock()
+	mq.tasks[t2.ID] = t2
+	mq.pendingTasks = append(mq.pendingTasks, t2)
+	mq.mutex.Unlock()
+	mq.AddTask(&MaintenanceTask{
+		ID:          "t3",
+		Type:        MaintenanceTaskType("vacuum"),
+		Priority:    PriorityNormal,
+		VolumeID:    200,
+		Server:      "server1",
+		ScheduledAt: now.Add(-1 * time.Second),
+	})
+
+	mq.workers["worker1"] = &MaintenanceWorker{
+		ID:            "worker1",
+		Status:        "active",
+		Capabilities:  []MaintenanceTaskType{"balance", "erasure_coding", "vacuum"},
+		MaxConcurrent: 2,
+		LastHeartbeat: time.Now(),
+	}
+	mq.workers["worker2"] = &MaintenanceWorker{
+		ID:            "worker2",
+		Status:        "active",
+		Capabilities:  []MaintenanceTaskType{"balance", "erasure_coding", "vacuum"},
+		MaxConcurrent: 2,
+		LastHeartbeat: time.Now(),
+	}
+
+	task1 := mq.GetNextTask("worker1", mq.workers["worker1"].Capabilities)
+	if task1 == nil || task1.ID != "t1" {
+		t.Fatalf("Expected first assignment to be t1, got %+v", task1)
+	}
+
+	task2 := mq.GetNextTask("worker2", mq.workers["worker2"].Capabilities)
+	if task2 == nil {
+		t.Fatalf("Expected a second task to be assigned, got nil")
+	}
+	if task2.ID != "t3" {
+		t.Fatalf("Expected second assignment to skip volume 100 and pick t3, got %s", task2.ID)
+	}
+
+	if mq.tasks["t2"].Status != TaskStatusPending {
+		t.Fatalf("Expected t2 to remain pending due to volume conflict, got %s", mq.tasks["t2"].Status)
+	}
+}
+
+func TestAddTask_OnePendingTaskPerVolume(t *testing.T) {
+	mq := NewMaintenanceQueue(&MaintenancePolicy{
+		TaskPolicies: map[string]*worker_pb.TaskPolicy{
+			"balance":        {MaxConcurrent: 1},
+			"erasure_coding": {MaxConcurrent: 1},
+		},
+	})
+
+	mq.AddTask(&MaintenanceTask{
+		ID:       "t1",
+		Type:     MaintenanceTaskType("balance"),
+		VolumeID: 100,
+		Server:   "server1",
+	})
+	mq.AddTask(&MaintenanceTask{
+		ID:       "t2",
+		Type:     MaintenanceTaskType("erasure_coding"),
+		VolumeID: 100,
+		Server:   "server1",
+	})
+
+	mq.mutex.RLock()
+	defer mq.mutex.RUnlock()
+
+	if len(mq.tasks) != 1 {
+		t.Fatalf("Expected 1 task in queue, got %d", len(mq.tasks))
+	}
+	if len(mq.pendingTasks) != 1 {
+		t.Fatalf("Expected 1 pending task, got %d", len(mq.pendingTasks))
+	}
+	if _, exists := mq.tasks["t1"]; !exists {
+		t.Fatalf("Expected task t1 to be queued")
+	}
+	if _, exists := mq.tasks["t2"]; exists {
+		t.Fatalf("Did not expect task t2 to be queued due to pending volume")
+	}
+}
+
+func TestAddTask_RejectsWhenVolumeHasRunningTask(t *testing.T) {
+	mq := NewMaintenanceQueue(&MaintenancePolicy{
+		TaskPolicies: map[string]*worker_pb.TaskPolicy{
+			"balance":        {MaxConcurrent: 1},
+			"erasure_coding": {MaxConcurrent: 1},
+		},
+	})
+
+	mq.AddTask(&MaintenanceTask{
+		ID:       "t1",
+		Type:     MaintenanceTaskType("balance"),
+		VolumeID: 100,
+		Server:   "server1",
+	})
+
+	// Simulate assignment to make it active
+	mq.mutex.Lock()
+	mq.tasks["t1"].Status = TaskStatusInProgress
+	mq.mutex.Unlock()
+
+	mq.AddTask(&MaintenanceTask{
+		ID:       "t2",
+		Type:     MaintenanceTaskType("erasure_coding"),
+		VolumeID: 100,
+		Server:   "server1",
+	})
+
+	mq.mutex.RLock()
+	defer mq.mutex.RUnlock()
+
+	if len(mq.tasks) != 1 {
+		t.Fatalf("Expected 1 task in queue, got %d", len(mq.tasks))
+	}
+	if _, exists := mq.tasks["t2"]; exists {
+		t.Fatalf("Did not expect task t2 to be queued due to active volume task")
+	}
+}

weed/admin/plugin/plugin.go

@@ -34,6 +34,13 @@ type Options struct {
 	ClusterContextProvider func(context.Context) (*plugin_pb.ClusterContext, error)
 }
 
+// JobTypeInfo contains metadata about a plugin job type.
+type JobTypeInfo struct {
+	JobType     string `json:"job_type"`
+	DisplayName string `json:"display_name"`
+	Weight      int32  `json:"weight"`
+}
+
 type Plugin struct {
 	plugin_pb.UnimplementedPluginControlServiceServer
 
@@ -623,7 +630,7 @@ func (r *Plugin) ListWorkers() []*WorkerSession {
 	return r.registry.List()
 }
 
-func (r *Plugin) ListKnownJobTypes() ([]string, error) {
+func (r *Plugin) ListKnownJobTypes() ([]JobTypeInfo, error) {
 	registryJobTypes := r.registry.JobTypes()
 	storedJobTypes, err := r.store.ListJobTypes()
 	if err != nil {
@@ -638,12 +645,72 @@ func (r *Plugin) ListKnownJobTypes() ([]string, error) {
 		jobTypeSet[jobType] = struct{}{}
 	}
 
-	out := make([]string, 0, len(jobTypeSet))
+	jobTypeList := make([]string, 0, len(jobTypeSet))
 	for jobType := range jobTypeSet {
-		out = append(out, jobType)
+		jobTypeList = append(jobTypeList, jobType)
 	}
-	sort.Strings(out)
-	return out, nil
+	sort.Strings(jobTypeList)
+
+	result := make([]JobTypeInfo, 0, len(jobTypeList))
+	workers := r.registry.List()
+
+	// Pre-calculate the best capability for each job type from available workers.
+	// Prefer capabilities with non-empty DisplayName, then higher Weight.
+	jobTypeToCap := make(map[string]*plugin_pb.JobTypeCapability)
+	for _, worker := range workers {
+		for jobType, cap := range worker.Capabilities {
+			if cap == nil {
+				continue
+			}
+			existing, exists := jobTypeToCap[jobType]
+			if !exists || existing == nil {
+				jobTypeToCap[jobType] = cap
+				continue
+			}
+			// Preserve existing if it has DisplayName but cap doesn't.
+			if existing.DisplayName != "" && cap.DisplayName == "" {
+				continue
+			}
+			// Prefer capabilities with a non-empty DisplayName.
+			if existing.DisplayName == "" && cap.DisplayName != "" {
+				jobTypeToCap[jobType] = cap
+				continue
+			}
+			// If DisplayName statuses are equal, prefer higher Weight.
+			if cap.Weight > existing.Weight {
+				jobTypeToCap[jobType] = cap
+			}
+		}
+	}
+
+	for _, jobType := range jobTypeList {
+		info := JobTypeInfo{JobType: jobType}
+
+		// Get display name and weight from pre-calculated capabilities
+		if cap, ok := jobTypeToCap[jobType]; ok && cap != nil {
+			if cap.DisplayName != "" {
+				info.DisplayName = cap.DisplayName
+			}
+			info.Weight = cap.Weight
+		}
+
+		// Default display name to job type if not set
+		if info.DisplayName == "" {
+			info.DisplayName = jobType
+		}
+
+		result = append(result, info)
+	}
+
+	// Sort by weight (descending) then by job type (ascending)
+	sort.Slice(result, func(i, j int) bool {
+		if result[i].Weight != result[j].Weight {
+			return result[i].Weight > result[j].Weight // higher weight first
+		}
+		return result[i].JobType < result[j].JobType // alphabetical as tiebreaker
+	})
+
+	return result, nil
 }
 
 // FilterProposalsWithActiveJobs drops proposals that are already assigned/running.

weed/admin/plugin/plugin_scheduler.go

@@ -175,7 +175,8 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) {
 	r.schedulerMu.Unlock()
 
 	states := make([]SchedulerJobTypeState, 0, len(jobTypes))
-	for _, jobType := range jobTypes {
+	for _, jobTypeInfo := range jobTypes {
+		jobType := jobTypeInfo.JobType
 		state := SchedulerJobTypeState{
 			JobType:           jobType,
 			DetectionInFlight: detectionInFlight[jobType],
@@ -187,6 +188,7 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) {
 		}
 
 		policy, enabled, loadErr := r.loadSchedulerPolicy(jobType)
+
 		if loadErr != nil {
 			state.PolicyError = loadErr.Error()
 		} else {

weed/admin/view/app/plugin.templ

@@ -683,6 +683,25 @@ templ Plugin(page string) {
                 return 'configuration';
             }
 
+            // Parse a job type item (string or object) into a safe object shape
+            function parseJobTypeItem(item) {
+                var jobType = '';
+                var displayName = '';
+                
+                if (typeof item === 'string') {
+                    jobType = String(item || '').trim();
+                    displayName = jobType;
+                } else if (item && typeof item === 'object') {
+                    jobType = String(item.job_type || item.jobType || '').trim();
+                    displayName = String(item.display_name || item.displayName || jobType || '').trim();
+                    if (!displayName) {
+                        displayName = jobType;
+                    }
+                }
+                
+                return { jobType: jobType, displayName: displayName };
+            }
+
             function topTabKeyForJobType(jobType) {
                 var normalized = String(jobType || '').trim();
                 if (!normalized) {
@@ -706,7 +725,8 @@ templ Plugin(page string) {
                 }
                 var jobTypes = Array.isArray(state.jobTypes) ? state.jobTypes : [];
                 for (var i = 0; i < jobTypes.length; i++) {
-                    if (String(jobTypes[i] || '').trim() === normalized) {
+                    var item = parseJobTypeItem(jobTypes[i]);
+                    if (item.jobType === normalized) {
                         return true;
                     }
                 }
@@ -728,7 +748,8 @@ templ Plugin(page string) {
                     state.activeTopTab = 'overview';
                     return;
                 }
-                state.selectedJobType = String(state.jobTypes[0] || '').trim();
+                var firstItem = parseJobTypeItem(state.jobTypes[0]);
+                state.selectedJobType = firstItem.jobType;
                 state.activeTopTab = topTabKeyForJobType(state.selectedJobType);
             }
 
@@ -749,7 +770,8 @@ templ Plugin(page string) {
 
                 if (state.activeTopTab === 'overview') {
                     if (!state.selectedJobType && Array.isArray(state.jobTypes) && state.jobTypes.length > 0) {
-                        state.selectedJobType = String(state.jobTypes[0] || '').trim();
+                        var item = parseJobTypeItem(state.jobTypes[0]);
+                        state.selectedJobType = item.jobType;
                     }
                     return;
                 }
@@ -758,7 +780,8 @@ templ Plugin(page string) {
                     topTabJobType = state.selectedJobType;
                 }
                 if (!topTabJobType && Array.isArray(state.jobTypes) && state.jobTypes.length > 0) {
-                    topTabJobType = String(state.jobTypes[0] || '').trim();
+                    var item = parseJobTypeItem(state.jobTypes[0]);
+                    topTabJobType = item.jobType;
                 }
                 if (!topTabJobType) {
                     state.activeTopTab = 'overview';
@@ -784,14 +807,17 @@ templ Plugin(page string) {
 
                 var jobTypes = Array.isArray(state.jobTypes) ? state.jobTypes : [];
                 for (var i = 0; i < jobTypes.length; i++) {
-                    var jobType = String(jobTypes[i] || '').trim();
+                    var item = parseJobTypeItem(jobTypes[i]);
+                    var jobType = item.jobType;
+                    var displayName = item.displayName;
+                    
                     if (!jobType) {
                         continue;
                     }
                     html += '' +
                         '<li class="nav-item">' +
                         '<button type="button" class="nav-link" data-plugin-top-tab="' + escapeHtml(topTabKeyForJobType(jobType)) + '">' +
-                        escapeHtml(jobType) +
+                        escapeHtml(displayName) +
                         '</button>' +
                         '</li>';
                 }
@@ -1267,7 +1293,8 @@ templ Plugin(page string) {
                 var jobTypes = [];
 
                 for (var k = 0; k < known.length; k++) {
-                    var knownType = String(known[k] || '').trim();
+                    var item = parseJobTypeItem(known[k]);
+                    var knownType = item && item.jobType ? String(item.jobType).trim() : '';
                     if (!knownType || seen[knownType]) {
                         continue;
                     }

weed/command/filer.go

@@ -145,6 +145,7 @@ func init() {
 	filerS3Options.cipher = cmdFiler.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads")
 	filerS3Options.iamReadOnly = cmdFiler.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server")
 	filerS3Options.portIceberg = cmdFiler.Flag.Int("s3.port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)")
+	filerS3Options.externalUrl = cmdFiler.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Used for S3 signature verification behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 
 	// start webdav on filer
 	filerStartWebDav = cmdFiler.Flag.Bool("webdav", false, "whether to start webdav gateway")

weed/command/mini.go

@@ -248,6 +248,7 @@ func initMiniS3Flags() {
 	miniS3Options.iamConfig = miniIamConfig
 	miniS3Options.auditLogConfig = cmdMini.Flag.String("s3.auditLogConfig", "", "path to the audit log config file")
 	miniS3Options.allowDeleteBucketNotEmpty = miniS3AllowDeleteBucketNotEmpty
+	miniS3Options.externalUrl = cmdMini.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Used for S3 signature verification behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 	// In mini mode, S3 uses the shared debug server started at line 681, not its own separate debug server
 	miniS3Options.debug = new(bool) // explicitly false
 	miniS3Options.debugPort = cmdMini.Flag.Int("s3.debug.port", 6060, "http port for debugging (unused in mini mode)")
@@ -727,6 +728,7 @@ func saveMiniConfiguration(dataFolder string) error {
 }
 
 func runMini(cmd *Command, args []string) bool {
+	*miniDataFolders = util.ResolvePath(*miniDataFolders)
 
 	// Capture which port flags were explicitly passed on CLI BEFORE config file is applied
 	// This is necessary to distinguish user-specified ports from defaults or config file options
@@ -1029,9 +1031,15 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) {
 		glog.Fatalf("Admin server readiness check failed: %v", err)
 	}
 
-	// Start worker after admin server is ready
-	startMiniWorker()
-	startMiniPluginWorker(ctx)
+	// Start consolidated worker runtime (both standard and plugin runtimes)
+	workerDir := filepath.Join(*miniDataFolders, "worker")
+	if err := os.MkdirAll(workerDir, 0755); err != nil {
+		glog.Fatalf("Failed to create unified worker directory: %v", err)
+	}
+
+	glog.Infof("Starting consolidated maintenance worker system (directory: %s)", workerDir)
+	startMiniWorker(workerDir)
+	startMiniPluginWorker(ctx, workerDir)
 
 	// Wait for worker to be ready by polling its gRPC port
 	workerGrpcAddr := fmt.Sprintf("%s:%d", bindIp, *miniAdminOptions.grpcPort)
@@ -1090,17 +1098,13 @@ func waitForWorkerReady(workerGrpcAddr string) {
 }
 
 // startMiniWorker starts a single worker for the admin server
-func startMiniWorker() {
-	glog.Infof("Starting maintenance worker for admin server")
+func startMiniWorker(workerDir string) {
+	glog.V(1).Infof("Initializing standard worker runtime")
 
 	adminAddr := fmt.Sprintf("%s:%d", *miniIp, *miniAdminOptions.port)
 	capabilities := "vacuum,ec,balance"
 
-	// Use worker directory under main data folder
-	workerDir := filepath.Join(*miniDataFolders, "worker")
-	if err := os.MkdirAll(workerDir, 0755); err != nil {
-		glog.Fatalf("Failed to create worker directory: %v", err)
-	}
+	// Use common worker directory
 
 	glog.Infof("Worker connecting to admin server: %s", adminAddr)
 	glog.Infof("Worker capabilities: %s", capabilities)
@@ -1169,7 +1173,7 @@ func startMiniWorker() {
 	glog.Infof("Maintenance worker %s started successfully", workerInstance.ID())
 }
 
-func startMiniPluginWorker(ctx context.Context) {
+func startMiniPluginWorker(ctx context.Context, workerDir string) {
 	glog.Infof("Starting plugin worker for admin server")
 
 	adminAddr := fmt.Sprintf("%s:%d", *miniIp, *miniAdminOptions.port)
@@ -1178,10 +1182,7 @@ func startMiniPluginWorker(ctx context.Context) {
 		glog.Infof("Resolved mini plugin worker admin endpoint: %s -> %s", adminAddr, resolvedAdminAddr)
 	}
 
-	workerDir := filepath.Join(*miniDataFolders, "plugin_worker")
-	if err := os.MkdirAll(workerDir, 0755); err != nil {
-		glog.Fatalf("Failed to create plugin worker directory: %v", err)
-	}
+	// Use common worker directory
 
 	util.LoadConfiguration("security", false)
 	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.worker")

weed/command/plugin_worker_test.go

@@ -153,11 +153,11 @@ func TestResolvePluginWorkerID(t *testing.T) {
 	if generated == "" {
 		t.Fatalf("expected generated id")
 	}
-	if len(generated) < 7 || generated[:7] != "plugin-" {
-		t.Fatalf("expected generated id prefix plugin-, got %q", generated)
+	if len(generated) < 2 || generated[:2] != "w-" {
+		t.Fatalf("expected generated id prefix w-, got %q", generated)
 	}
 
-	persistedPath := filepath.Join(dir, "plugin.worker.id")
+	persistedPath := filepath.Join(dir, "worker.id")
 	if _, statErr := os.Stat(persistedPath); statErr != nil {
 		t.Fatalf("expected persisted worker id file: %v", statErr)
 	}

weed/command/s3.go

@@ -67,6 +67,7 @@ type S3Options struct {
 	debug                     *bool
 	debugPort                 *int
 	cipher                    *bool
+	externalUrl               *string
 }
 
 func init() {
@@ -101,6 +102,7 @@ func init() {
 	s3StandaloneOptions.debug = cmdS3.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
 	s3StandaloneOptions.debugPort = cmdS3.Flag.Int("debug.port", 6060, "http port for debugging")
 	s3StandaloneOptions.cipher = cmdS3.Flag.Bool("encryptVolumeData", false, "encrypt data on volume servers")
+	s3StandaloneOptions.externalUrl = cmdS3.Flag.String("externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Used for S3 signature verification behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 }
 
 var cmdS3 = &Command{
@@ -222,6 +224,14 @@ func (s3opt *S3Options) GetCertificateWithUpdate(*tls.ClientHelloInfo) (*tls.Cer
 	return &certs.Certs[0], err
 }
 
+// resolveExternalUrl returns the external URL from the flag or falls back to the S3_EXTERNAL_URL env var.
+func (s3opt *S3Options) resolveExternalUrl() string {
+	if s3opt.externalUrl != nil && *s3opt.externalUrl != "" {
+		return *s3opt.externalUrl
+	}
+	return os.Getenv("S3_EXTERNAL_URL")
+}
+
 func (s3opt *S3Options) startS3Server() bool {
 
 	filerAddresses := pb.ServerAddresses(*s3opt.filer).ToAddresses()
@@ -309,6 +319,7 @@ func (s3opt *S3Options) startS3Server() bool {
 		Cipher:                    *s3opt.cipher, // encrypt data on volume servers
 		BindIp:                    *s3opt.bindIp,
 		GrpcPort:                  *s3opt.portGrpc,
+		ExternalUrl:               s3opt.resolveExternalUrl(),
 	})
 	if s3ApiServer_err != nil {
 		glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)

weed/command/server.go

@@ -178,6 +178,7 @@ func init() {
 	s3Options.enableIam = cmdServer.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same S3 port")
 	s3Options.iamReadOnly = cmdServer.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server")
 	s3Options.cipher = cmdServer.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads")
+	s3Options.externalUrl = cmdServer.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Used for S3 signature verification behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 
 	sftpOptions.port = cmdServer.Flag.Int("sftp.port", 2022, "SFTP server listen port")
 	sftpOptions.sshPrivateKey = cmdServer.Flag.String("sftp.sshPrivateKey", "", "path to the SSH private key file for host authentication")

weed/command/worker.go

@@ -17,7 +17,7 @@ heartbeat/load reporting, detection, and execution.
 
 Behavior:
   - Use -jobType to choose one or more plugin job handlers (comma-separated list)
-  - Use -workingDir to persist plugin.worker.id for stable worker identity across restarts
+  - Use -workingDir to persist worker.id for stable worker identity across restarts
   - Use -metricsPort/-metricsIp to expose /health, /ready, and /metrics
 
 Examples:

weed/command/worker_runtime.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"path/filepath"
 	"strconv"
 	"strings"
 	"syscall"
@@ -20,6 +19,7 @@ import (
 	statsCollect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/version"
+	"github.com/seaweedfs/seaweedfs/weed/worker"
 	"google.golang.org/grpc"
 )
 
@@ -130,31 +130,11 @@ func runPluginWorkerWithOptions(options pluginWorkerRunOptions) bool {
 }
 
 func resolvePluginWorkerID(explicitID string, workingDir string) (string, error) {
-	id := strings.TrimSpace(explicitID)
-	if id != "" {
-		return id, nil
+	if explicitID != "" {
+		return explicitID, nil
 	}
-
-	workingDir = strings.TrimSpace(workingDir)
-	if workingDir == "" {
-		return "", nil
-	}
-	if err := os.MkdirAll(workingDir, 0755); err != nil {
-		return "", err
-	}
-
-	workerIDPath := filepath.Join(workingDir, "plugin.worker.id")
-	if data, err := os.ReadFile(workerIDPath); err == nil {
-		if persisted := strings.TrimSpace(string(data)); persisted != "" {
-			return persisted, nil
-		}
-	}
-
-	generated := fmt.Sprintf("plugin-%d", time.Now().UnixNano())
-	if err := os.WriteFile(workerIDPath, []byte(generated+"\n"), 0644); err != nil {
-		return "", err
-	}
-	return generated, nil
+	// Use the same ID generation/loading logic as the standard worker
+	return worker.GenerateOrLoadWorkerID(workingDir)
 }
 
 // buildPluginWorkerHandler constructs the JobHandler for the given job type.

weed/pb/plugin.proto

@@ -96,6 +96,7 @@ message JobTypeCapability {
   int32 max_execution_concurrency = 5;
   string display_name = 6;
   string description = 7;
+  int32 weight = 8;
 }
 
 message RequestConfigSchema {

weed/pb/plugin_pb/plugin.pb.go

@@ -1209,6 +1209,7 @@ type JobTypeCapability struct {
 	MaxExecutionConcurrency int32                  `protobuf:"varint,5,opt,name=max_execution_concurrency,json=maxExecutionConcurrency,proto3" json:"max_execution_concurrency,omitempty"`
 	DisplayName             string                 `protobuf:"bytes,6,opt,name=display_name,json=displayName,proto3" json:"display_name,omitempty"`
 	Description             string                 `protobuf:"bytes,7,opt,name=description,proto3" json:"description,omitempty"`
+	Weight                  int32                  `protobuf:"varint,8,opt,name=weight,proto3" json:"weight,omitempty"`
 	unknownFields           protoimpl.UnknownFields
 	sizeCache               protoimpl.SizeCache
 }
@@ -1292,6 +1293,13 @@ func (x *JobTypeCapability) GetDescription() string {
 	return ""
 }
 
+func (x *JobTypeCapability) GetWeight() int32 {
+	if x != nil {
+		return x.Weight
+	}
+	return 0
+}
+
 type RequestConfigSchema struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	JobType       string                 `protobuf:"bytes,1,opt,name=job_type,json=jobType,proto3" json:"job_type,omitempty"`
@@ -3948,7 +3956,7 @@ const file_plugin_proto_rawDesc = "" +
 	"\bjob_type\x18\x03 \x01(\tR\ajobType\x12&\n" +
 	"\x05state\x18\x04 \x01(\x0e2\x10.plugin.JobStateR\x05state\x12)\n" +
 	"\x10progress_percent\x18\x05 \x01(\x01R\x0fprogressPercent\x12\x14\n" +
-	"\x05stage\x18\x06 \x01(\tR\x05stage\"\xab\x02\n" +
+	"\x05stage\x18\x06 \x01(\tR\x05stage\"\xc3\x02\n" +
 	"\x11JobTypeCapability\x12\x19\n" +
 	"\bjob_type\x18\x01 \x01(\tR\ajobType\x12\x1d\n" +
 	"\n" +
@@ -3958,7 +3966,8 @@ const file_plugin_proto_rawDesc = "" +
 	"\x19max_detection_concurrency\x18\x04 \x01(\x05R\x17maxDetectionConcurrency\x12:\n" +
 	"\x19max_execution_concurrency\x18\x05 \x01(\x05R\x17maxExecutionConcurrency\x12!\n" +
 	"\fdisplay_name\x18\x06 \x01(\tR\vdisplayName\x12 \n" +
-	"\vdescription\x18\a \x01(\tR\vdescription\"U\n" +
+	"\vdescription\x18\a \x01(\tR\vdescription\x12\x16\n" +
+	"\x06weight\x18\b \x01(\x05R\x06weight\"U\n" +
 	"\x13RequestConfigSchema\x12\x19\n" +
 	"\bjob_type\x18\x01 \x01(\tR\ajobType\x12#\n" +
 	"\rforce_refresh\x18\x02 \x01(\bR\fforceRefresh\"\xda\x01\n" +

weed/plugin/worker/erasure_coding_handler.go

@@ -12,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	ecstorage "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	erasurecodingtask "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding"
 	workertypes "github.com/seaweedfs/seaweedfs/weed/worker/types"
 	"google.golang.org/grpc"
@@ -40,15 +41,16 @@ func (h *ErasureCodingHandler) Capability() *plugin_pb.JobTypeCapability {
 		CanExecute:              true,
 		MaxDetectionConcurrency: 1,
 		MaxExecutionConcurrency: 1,
-		DisplayName:             "Erasure Coding",
+		DisplayName:             "EC Encoding",
 		Description:             "Converts full and quiet volumes into EC shards",
+		Weight:                  80,
 	}
 }
 
 func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor {
 	return &plugin_pb.JobTypeDescriptor{
 		JobType:           "erasure_coding",
-		DisplayName:       "Erasure Coding",
+		DisplayName:       "EC Encoding",
 		Description:       "Detect and execute erasure coding for suitable volumes",
 		Icon:              "fas fa-shield-alt",
 		DescriptorVersion: 1,
@@ -279,7 +281,7 @@ func emitErasureCodingDetectionDecisionTrace(
 
 	quietThreshold := time.Duration(taskConfig.QuietForSeconds) * time.Second
 	minSizeBytes := uint64(taskConfig.MinSizeMB) * 1024 * 1024
-	allowedCollections := erasurecodingtask.ParseCollectionFilter(taskConfig.CollectionFilter)
+	allowedCollections := wildcard.CompileWildcardMatchers(taskConfig.CollectionFilter)
 
 	volumeGroups := make(map[uint32][]*workertypes.VolumeHealthMetrics)
 	for _, metric := range metrics {
@@ -317,7 +319,7 @@ func emitErasureCodingDetectionDecisionTrace(
 			skippedTooSmall++
 			continue
 		}
-		if len(allowedCollections) > 0 && !allowedCollections[metric.Collection] {
+		if len(allowedCollections) > 0 && !wildcard.MatchesAnyWildcard(allowedCollections, metric.Collection) {
 			skippedCollectionFilter++
 			continue
 		}
@@ -568,9 +570,7 @@ func (h *ErasureCodingHandler) collectVolumeMetrics(
 	masterAddresses []string,
 	collectionFilter string,
 ) ([]*workertypes.VolumeHealthMetrics, *topology.ActiveTopology, error) {
-	// Reuse the same master topology fetch/build flow used by the vacuum handler.
-	helper := &VacuumHandler{grpcDialOption: h.grpcDialOption}
-	return helper.collectVolumeMetrics(ctx, masterAddresses, collectionFilter)
+	return collectVolumeMetricsFromMasters(ctx, masterAddresses, collectionFilter, h.grpcDialOption)
 }
 
 func deriveErasureCodingWorkerConfig(values map[string]*plugin_pb.ConfigValue) *erasureCodingWorkerConfig {
@@ -905,7 +905,7 @@ func assignECShardIDs(totalShards int, targetCount int) [][]uint32 {
 func defaultErasureCodingWorkingDir(baseWorkingDir string) string {
 	dir := strings.TrimSpace(baseWorkingDir)
 	if dir == "" {
-		return filepath.Join(".", "seaweedfs-ec")
+		return filepath.Join(".", "erasure_coding")
 	}
-	return filepath.Join(dir, "seaweedfs-ec")
+	return filepath.Join(dir, "erasure_coding")
 }

weed/plugin/worker/vacuum_handler.go

@@ -3,15 +3,12 @@ package pluginworker
 import (
 	"context"
 	"fmt"
-	"sort"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	vacuumtask "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
@@ -55,6 +52,7 @@ func (h *VacuumHandler) Capability() *plugin_pb.JobTypeCapability {
 		MaxExecutionConcurrency: maxExec,
 		DisplayName:             "Volume Vacuum",
 		Description:             "Reclaims disk space by removing deleted files from volumes",
+		Weight:                  60,
 	}
 }
 
@@ -491,62 +489,7 @@ func (h *VacuumHandler) collectVolumeMetrics(
 	masterAddresses []string,
 	collectionFilter string,
 ) ([]*workertypes.VolumeHealthMetrics, *topology.ActiveTopology, error) {
-	if h.grpcDialOption == nil {
-		return nil, nil, fmt.Errorf("grpc dial option is not configured")
-	}
-	if len(masterAddresses) == 0 {
-		return nil, nil, fmt.Errorf("no master addresses provided in cluster context")
-	}
-
-	for _, masterAddress := range masterAddresses {
-		response, err := h.fetchVolumeList(ctx, masterAddress)
-		if err != nil {
-			glog.Warningf("Plugin worker failed master volume list at %s: %v", masterAddress, err)
-			continue
-		}
-
-		metrics, activeTopology, buildErr := buildVolumeMetrics(response, collectionFilter)
-		if buildErr != nil {
-			glog.Warningf("Plugin worker failed to build metrics from master %s: %v", masterAddress, buildErr)
-			continue
-		}
-		return metrics, activeTopology, nil
-	}
-
-	return nil, nil, fmt.Errorf("failed to load topology from all provided masters")
-}
-
-func (h *VacuumHandler) fetchVolumeList(ctx context.Context, address string) (*master_pb.VolumeListResponse, error) {
-	var lastErr error
-	for _, candidate := range masterAddressCandidates(address) {
-		if ctx.Err() != nil {
-			return nil, ctx.Err()
-		}
-
-		dialCtx, cancelDial := context.WithTimeout(ctx, 5*time.Second)
-		conn, err := pb.GrpcDial(dialCtx, candidate, false, h.grpcDialOption)
-		cancelDial()
-		if err != nil {
-			lastErr = err
-			continue
-		}
-
-		client := master_pb.NewSeaweedClient(conn)
-		callCtx, cancelCall := context.WithTimeout(ctx, 10*time.Second)
-		response, callErr := client.VolumeList(callCtx, &master_pb.VolumeListRequest{})
-		cancelCall()
-		_ = conn.Close()
-
-		if callErr == nil {
-			return response, nil
-		}
-		lastErr = callErr
-	}
-
-	if lastErr == nil {
-		lastErr = fmt.Errorf("no valid master address candidate")
-	}
-	return nil, lastErr
+	return collectVolumeMetricsFromMasters(ctx, masterAddresses, collectionFilter, h.grpcDialOption)
 }
 
 func deriveVacuumConfig(values map[string]*plugin_pb.ConfigValue) *vacuumtask.Config {
@@ -557,74 +500,6 @@ func deriveVacuumConfig(values map[string]*plugin_pb.ConfigValue) *vacuumtask.Co
 	return config
 }
 
-func buildVolumeMetrics(
-	response *master_pb.VolumeListResponse,
-	collectionFilter string,
-) ([]*workertypes.VolumeHealthMetrics, *topology.ActiveTopology, error) {
-	if response == nil || response.TopologyInfo == nil {
-		return nil, nil, fmt.Errorf("volume list response has no topology info")
-	}
-
-	activeTopology := topology.NewActiveTopology(10)
-	if err := activeTopology.UpdateTopology(response.TopologyInfo); err != nil {
-		return nil, nil, err
-	}
-
-	filter := strings.TrimSpace(collectionFilter)
-	volumeSizeLimitBytes := uint64(response.VolumeSizeLimitMb) * 1024 * 1024
-	now := time.Now()
-	metrics := make([]*workertypes.VolumeHealthMetrics, 0, 256)
-
-	for _, dc := range response.TopologyInfo.DataCenterInfos {
-		for _, rack := range dc.RackInfos {
-			for _, node := range rack.DataNodeInfos {
-				for diskType, diskInfo := range node.DiskInfos {
-					for _, volume := range diskInfo.VolumeInfos {
-						if filter != "" && volume.Collection != filter {
-							continue
-						}
-
-						metric := &workertypes.VolumeHealthMetrics{
-							VolumeID:         volume.Id,
-							Server:           node.Id,
-							ServerAddress:    string(pb.NewServerAddressFromDataNode(node)),
-							DiskType:         diskType,
-							DiskId:           volume.DiskId,
-							DataCenter:       dc.Id,
-							Rack:             rack.Id,
-							Collection:       volume.Collection,
-							Size:             volume.Size,
-							DeletedBytes:     volume.DeletedByteCount,
-							LastModified:     time.Unix(volume.ModifiedAtSecond, 0),
-							ReplicaCount:     1,
-							ExpectedReplicas: int(volume.ReplicaPlacement),
-							IsReadOnly:       volume.ReadOnly,
-						}
-						if metric.Size > 0 {
-							metric.GarbageRatio = float64(metric.DeletedBytes) / float64(metric.Size)
-						}
-						if volumeSizeLimitBytes > 0 {
-							metric.FullnessRatio = float64(metric.Size) / float64(volumeSizeLimitBytes)
-						}
-						metric.Age = now.Sub(metric.LastModified)
-						metrics = append(metrics, metric)
-					}
-				}
-			}
-		}
-	}
-
-	replicaCounts := make(map[uint32]int)
-	for _, metric := range metrics {
-		replicaCounts[metric.VolumeID]++
-	}
-	for _, metric := range metrics {
-		metric.ReplicaCount = replicaCounts[metric.VolumeID]
-	}
-
-	return metrics, activeTopology, nil
-}
-
 func buildVacuumProposal(result *workertypes.TaskDetectionResult) (*plugin_pb.JobProposal, error) {
 	if result == nil {
 		return nil, fmt.Errorf("task detection result is nil")
@@ -833,25 +708,6 @@ func mapTaskPriority(priority workertypes.TaskPriority) plugin_pb.JobPriority {
 	}
 }
 
-func masterAddressCandidates(address string) []string {
-	trimmed := strings.TrimSpace(address)
-	if trimmed == "" {
-		return nil
-	}
-	candidateSet := map[string]struct{}{
-		trimmed: {},
-	}
-	converted := pb.ServerToGrpcAddress(trimmed)
-	candidateSet[converted] = struct{}{}
-
-	candidates := make([]string, 0, len(candidateSet))
-	for candidate := range candidateSet {
-		candidates = append(candidates, candidate)
-	}
-	sort.Strings(candidates)
-	return candidates
-}
-
 func shouldSkipDetectionByInterval(lastSuccessfulRun *timestamppb.Timestamp, minIntervalSeconds int) bool {
 	if lastSuccessfulRun == nil || minIntervalSeconds <= 0 {
 		return false

weed/plugin/worker/volume_balance_handler.go

@@ -44,6 +44,7 @@ func (h *VolumeBalanceHandler) Capability() *plugin_pb.JobTypeCapability {
 		MaxExecutionConcurrency: 1,
 		DisplayName:             "Volume Balance",
 		Description:             "Moves volumes between servers to reduce skew in volume distribution",
+		Weight:                  50,
 	}
 }
 

weed/plugin/worker/volume_metrics.go

@@ -0,0 +1,168 @@
+package pluginworker
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
+	workertypes "github.com/seaweedfs/seaweedfs/weed/worker/types"
+	"google.golang.org/grpc"
+)
+
+func collectVolumeMetricsFromMasters(
+	ctx context.Context,
+	masterAddresses []string,
+	collectionFilter string,
+	grpcDialOption grpc.DialOption,
+) ([]*workertypes.VolumeHealthMetrics, *topology.ActiveTopology, error) {
+	if grpcDialOption == nil {
+		return nil, nil, fmt.Errorf("grpc dial option is not configured")
+	}
+	if len(masterAddresses) == 0 {
+		return nil, nil, fmt.Errorf("no master addresses provided in cluster context")
+	}
+
+	for _, masterAddress := range masterAddresses {
+		response, err := fetchVolumeList(ctx, masterAddress, grpcDialOption)
+		if err != nil {
+			glog.Warningf("Plugin worker failed master volume list at %s: %v", masterAddress, err)
+			continue
+		}
+
+		metrics, activeTopology, buildErr := buildVolumeMetrics(response, collectionFilter)
+		if buildErr != nil {
+			glog.Warningf("Plugin worker failed to build metrics from master %s: %v", masterAddress, buildErr)
+			continue
+		}
+		return metrics, activeTopology, nil
+	}
+
+	return nil, nil, fmt.Errorf("failed to load topology from all provided masters")
+}
+
+func fetchVolumeList(ctx context.Context, address string, grpcDialOption grpc.DialOption) (*master_pb.VolumeListResponse, error) {
+	var lastErr error
+	for _, candidate := range masterAddressCandidates(address) {
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
+
+		dialCtx, cancelDial := context.WithTimeout(ctx, 5*time.Second)
+		conn, err := pb.GrpcDial(dialCtx, candidate, false, grpcDialOption)
+		cancelDial()
+		if err != nil {
+			lastErr = err
+			continue
+		}
+
+		client := master_pb.NewSeaweedClient(conn)
+		callCtx, cancelCall := context.WithTimeout(ctx, 10*time.Second)
+		response, callErr := client.VolumeList(callCtx, &master_pb.VolumeListRequest{})
+		cancelCall()
+		_ = conn.Close()
+
+		if callErr == nil {
+			return response, nil
+		}
+		lastErr = callErr
+	}
+
+	if lastErr == nil {
+		lastErr = fmt.Errorf("no valid master address candidate")
+	}
+	return nil, lastErr
+}
+
+func buildVolumeMetrics(
+	response *master_pb.VolumeListResponse,
+	collectionFilter string,
+) ([]*workertypes.VolumeHealthMetrics, *topology.ActiveTopology, error) {
+	if response == nil || response.TopologyInfo == nil {
+		return nil, nil, fmt.Errorf("volume list response has no topology info")
+	}
+
+	activeTopology := topology.NewActiveTopology(10)
+	if err := activeTopology.UpdateTopology(response.TopologyInfo); err != nil {
+		return nil, nil, err
+	}
+
+	patterns := wildcard.CompileWildcardMatchers(collectionFilter)
+	volumeSizeLimitBytes := uint64(response.VolumeSizeLimitMb) * 1024 * 1024
+	now := time.Now()
+	metrics := make([]*workertypes.VolumeHealthMetrics, 0, 256)
+
+	for _, dc := range response.TopologyInfo.DataCenterInfos {
+		for _, rack := range dc.RackInfos {
+			for _, node := range rack.DataNodeInfos {
+				for diskType, diskInfo := range node.DiskInfos {
+					for _, volume := range diskInfo.VolumeInfos {
+						if !wildcard.MatchesAnyWildcard(patterns, volume.Collection) {
+							continue
+						}
+
+						metric := &workertypes.VolumeHealthMetrics{
+							VolumeID:         volume.Id,
+							Server:           node.Id,
+							ServerAddress:    string(pb.NewServerAddressFromDataNode(node)),
+							DiskType:         diskType,
+							DiskId:           volume.DiskId,
+							DataCenter:       dc.Id,
+							Rack:             rack.Id,
+							Collection:       volume.Collection,
+							Size:             volume.Size,
+							DeletedBytes:     volume.DeletedByteCount,
+							LastModified:     time.Unix(volume.ModifiedAtSecond, 0),
+							ReplicaCount:     1,
+							ExpectedReplicas: int(volume.ReplicaPlacement),
+							IsReadOnly:       volume.ReadOnly,
+						}
+						if metric.Size > 0 {
+							metric.GarbageRatio = float64(metric.DeletedBytes) / float64(metric.Size)
+						}
+						if volumeSizeLimitBytes > 0 {
+							metric.FullnessRatio = float64(metric.Size) / float64(volumeSizeLimitBytes)
+						}
+						metric.Age = now.Sub(metric.LastModified)
+						metrics = append(metrics, metric)
+					}
+				}
+			}
+		}
+	}
+
+	replicaCounts := make(map[uint32]int)
+	for _, metric := range metrics {
+		replicaCounts[metric.VolumeID]++
+	}
+	for _, metric := range metrics {
+		metric.ReplicaCount = replicaCounts[metric.VolumeID]
+	}
+
+	return metrics, activeTopology, nil
+}
+
+func masterAddressCandidates(address string) []string {
+	trimmed := strings.TrimSpace(address)
+	if trimmed == "" {
+		return nil
+	}
+	candidateSet := map[string]struct{}{
+		trimmed: {},
+	}
+	converted := pb.ServerToGrpcAddress(trimmed)
+	candidateSet[converted] = struct{}{}
+
+	candidates := make([]string, 0, len(candidateSet))
+	for candidate := range candidateSet {
+		candidates = append(candidates, candidate)
+	}
+	sort.Strings(candidates)
+	return candidates
+}

weed/s3api/auth_credentials.go

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"slices"
 	"strings"
@@ -19,9 +21,9 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 
 	// Import KMS providers to register them
@@ -53,6 +55,7 @@ type IdentityAccessManagement struct {
 	identityAnonymous *Identity
 	hashMu            sync.RWMutex
 	domain            string
+	externalHost      string // pre-computed host for S3 signature verification (from ExternalUrl)
 	isAuthEnabled     bool
 	credentialManager *credential.CredentialManager
 	filerClient       *wdclient.FilerClient
@@ -154,13 +157,56 @@ func (iam *IdentityAccessManagement) SetFilerClient(filerClient *wdclient.FilerC
 	}
 }
 
+// parseExternalUrlToHost parses an external URL and returns the host string
+// to use for S3 signature verification. It applies the same default port
+// stripping rules as the AWS SDK: port 80 is stripped for HTTP, port 443
+// is stripped for HTTPS, all other ports are preserved.
+// Returns empty string for empty input.
+func parseExternalUrlToHost(externalUrl string) (string, error) {
+	if externalUrl == "" {
+		return "", nil
+	}
+	u, err := url.Parse(externalUrl)
+	if err != nil {
+		return "", fmt.Errorf("invalid external URL: parse failed")
+	}
+	if u.Host == "" {
+		return "", fmt.Errorf("invalid external URL: missing host")
+	}
+	host, port, err := net.SplitHostPort(u.Host)
+	if err != nil {
+		// No port in the URL. For IPv6, strip brackets to match AWS SDK.
+		if strings.Contains(u.Host, ":") {
+			return strings.Trim(u.Host, "[]"), nil
+		}
+		return u.Host, nil
+	}
+	// Strip default ports to match AWS SDK SanitizeHostForHeader behavior
+	if (port == "80" && strings.EqualFold(u.Scheme, "http")) ||
+		(port == "443" && strings.EqualFold(u.Scheme, "https")) {
+		return host, nil
+	}
+	return net.JoinHostPort(host, port), nil
+}
+
 func NewIdentityAccessManagement(option *S3ApiServerOption, filerClient *wdclient.FilerClient) *IdentityAccessManagement {
 	return NewIdentityAccessManagementWithStore(option, filerClient, "")
 }
 
 func NewIdentityAccessManagementWithStore(option *S3ApiServerOption, filerClient *wdclient.FilerClient, explicitStore string) *IdentityAccessManagement {
+	var externalHost string
+	if option.ExternalUrl != "" {
+		var err error
+		externalHost, err = parseExternalUrlToHost(option.ExternalUrl)
+		if err != nil {
+			glog.Fatalf("failed to parse s3.externalUrl: %v", err)
+		}
+		glog.V(0).Infof("S3 signature verification will use external host: %q (from %q)", externalHost, option.ExternalUrl)
+	}
+
 	iam := &IdentityAccessManagement{
 		domain:       option.DomainName,
+		externalHost: externalHost,
 		hashes:       make(map[string]*sync.Pool),
 		hashCounters: make(map[string]*int32),
 		filerClient:  filerClient,
@@ -1425,10 +1471,10 @@ func (identity *Identity) CanDo(action Action, bucket string, objectKey string)
 		act := string(a)
 		if strings.ContainsAny(act, "*?") {
 			// Pattern has wildcards - use smart matching
-			if policy_engine.MatchesWildcard(act, target) {
+			if wildcard.MatchesWildcard(act, target) {
 				return true
 			}
-			if policy_engine.MatchesWildcard(act, adminTarget) {
+			if wildcard.MatchesWildcard(act, adminTarget) {
 				return true
 			}
 		} else {

weed/s3api/auth_credentials_test.go

@@ -8,8 +8,8 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/credential"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
@@ -252,9 +252,9 @@ func TestMatchWildcardPattern(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.pattern+"_"+tt.target, func(t *testing.T) {
-			result := policy_engine.MatchesWildcard(tt.pattern, tt.target)
+			result := wildcard.MatchesWildcard(tt.pattern, tt.target)
 			if result != tt.match {
-				t.Errorf("policy_engine.MatchesWildcard(%q, %q) = %v, want %v", tt.pattern, tt.target, result, tt.match)
+				t.Errorf("wildcard.MatchesWildcard(%q, %q) = %v, want %v", tt.pattern, tt.target, result, tt.match)
 			}
 		})
 	}
@@ -805,3 +805,85 @@ func TestStaticIdentityProtection(t *testing.T) {
 	iam.m.RUnlock()
 	assert.False(t, ok, "Dynamic identity should have been removed")
 }
+
+func TestParseExternalUrlToHost(t *testing.T) {
+	tests := []struct {
+		name      string
+		input     string
+		expected  string
+		expectErr bool
+	}{
+		{
+			name:     "empty string",
+			input:    "",
+			expected: "",
+		},
+		{
+			name:     "HTTPS with default port stripped",
+			input:    "https://api.example.com:443",
+			expected: "api.example.com",
+		},
+		{
+			name:     "HTTP with default port stripped",
+			input:    "http://api.example.com:80",
+			expected: "api.example.com",
+		},
+		{
+			name:     "HTTPS with non-standard port preserved",
+			input:    "https://api.example.com:9000",
+			expected: "api.example.com:9000",
+		},
+		{
+			name:     "HTTP with non-standard port preserved",
+			input:    "http://api.example.com:8080",
+			expected: "api.example.com:8080",
+		},
+		{
+			name:     "HTTPS without port",
+			input:    "https://api.example.com",
+			expected: "api.example.com",
+		},
+		{
+			name:     "HTTP without port",
+			input:    "http://api.example.com",
+			expected: "api.example.com",
+		},
+		{
+			name:     "IPv6 with non-standard port",
+			input:    "https://[::1]:9000",
+			expected: "[::1]:9000",
+		},
+		{
+			name:     "IPv6 with default HTTPS port stripped",
+			input:    "https://[::1]:443",
+			expected: "::1",
+		},
+		{
+			name:     "IPv6 without port",
+			input:    "https://[::1]",
+			expected: "::1",
+		},
+		{
+			name:      "invalid URL",
+			input:     "://not-a-url",
+			expectErr: true,
+		},
+		{
+			name:      "missing host",
+			input:     "https://",
+			expectErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := parseExternalUrlToHost(tt.input)
+			if tt.expectErr {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

weed/s3api/auth_proxy_integration_test.go

@@ -0,0 +1,212 @@
+package s3api
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestReverseProxySignatureVerification is an integration test that exercises
+// the full HTTP stack: real AWS SDK v4 signer -> real httputil.ReverseProxy ->
+// real net/http server running IAM signature verification.
+//
+// This catches issues that unit tests miss: header normalization by net/http,
+// proxy header injection, and real-world Host header handling.
+func TestReverseProxySignatureVerification(t *testing.T) {
+	const (
+		accessKey = "AKIAIOSFODNN7EXAMPLE"
+		secretKey = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+	)
+
+	configJSON := `{
+  "identities": [
+    {
+      "name": "test_user",
+      "credentials": [
+        {
+          "accessKey": "` + accessKey + `",
+          "secretKey": "` + secretKey + `"
+        }
+      ],
+      "actions": ["Admin", "Read", "Write", "List", "Tagging"]
+    }
+  ]
+}`
+
+	tests := []struct {
+		name              string
+		externalUrl       string // s3.externalUrl config for the backend
+		clientScheme      string // scheme the client uses for signing
+		clientHost        string // host the client signs against
+		proxyForwardsHost bool   // whether proxy sets X-Forwarded-Host
+		expectSuccess     bool
+	}{
+		{
+			name:              "non-standard port, externalUrl matches proxy address",
+			externalUrl:       "", // filled dynamically with proxy address
+			clientScheme:      "http",
+			clientHost:        "", // filled dynamically
+			proxyForwardsHost: true,
+			expectSuccess:     true,
+		},
+		{
+			name:              "externalUrl with non-standard port, client signs against external host",
+			externalUrl:       "http://api.example.com:9000",
+			clientScheme:      "http",
+			clientHost:        "api.example.com:9000",
+			proxyForwardsHost: true,
+			expectSuccess:     true,
+		},
+		{
+			name:              "externalUrl with HTTPS default port stripped, client signs without port",
+			externalUrl:       "https://api.example.com:443",
+			clientScheme:      "https",
+			clientHost:        "api.example.com",
+			proxyForwardsHost: true,
+			expectSuccess:     true,
+		},
+		{
+			name:              "externalUrl with HTTP default port stripped, client signs without port",
+			externalUrl:       "http://api.example.com:80",
+			clientScheme:      "http",
+			clientHost:        "api.example.com",
+			proxyForwardsHost: true,
+			expectSuccess:     true,
+		},
+		{
+			name:              "proxy forwards X-Forwarded-Host correctly, no externalUrl needed",
+			externalUrl:       "",
+			clientScheme:      "http",
+			clientHost:        "api.example.com:9000",
+			proxyForwardsHost: true,
+			expectSuccess:     true,
+		},
+		{
+			name:              "proxy without X-Forwarded-Host, no externalUrl: host mismatch",
+			externalUrl:       "",
+			clientScheme:      "http",
+			clientHost:        "api.example.com:9000",
+			proxyForwardsHost: false,
+			expectSuccess:     false,
+		},
+		{
+			name:              "proxy without X-Forwarded-Host, externalUrl saves the day",
+			externalUrl:       "http://api.example.com:9000",
+			clientScheme:      "http",
+			clientHost:        "api.example.com:9000",
+			proxyForwardsHost: false,
+			expectSuccess:     true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// --- Write config to temp file ---
+			tmpFile := t.TempDir() + "/s3.json"
+			require.NoError(t, os.WriteFile(tmpFile, []byte(configJSON), 0644))
+
+			// --- Set up backend ---
+			var iam *IdentityAccessManagement
+			backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				_, errCode := iam.authRequest(r, "Read")
+				if errCode != 0 {
+					w.WriteHeader(http.StatusForbidden)
+					fmt.Fprintf(w, "error: %d", int(errCode))
+					return
+				}
+				w.WriteHeader(http.StatusOK)
+				fmt.Fprint(w, "OK")
+			}))
+			defer backend.Close()
+
+			// --- Set up reverse proxy ---
+			backendURL, _ := url.Parse(backend.URL)
+			proxy := httputil.NewSingleHostReverseProxy(backendURL)
+
+			forwardsHost := tt.proxyForwardsHost
+			originalDirector := proxy.Director
+			proxy.Director = func(req *http.Request) {
+				originalHost := req.Host
+				originalScheme := req.URL.Scheme
+				if originalScheme == "" {
+					originalScheme = "http"
+				}
+				originalDirector(req)
+				// Simulate real proxy behavior: rewrite Host to backend address
+				// (nginx proxy_pass and Kong both do this by default)
+				req.Host = backendURL.Host
+				if forwardsHost {
+					req.Header.Set("X-Forwarded-Host", originalHost)
+					req.Header.Set("X-Forwarded-Proto", originalScheme)
+				}
+			}
+
+			proxyServer := httptest.NewServer(proxy)
+			defer proxyServer.Close()
+
+			// --- Configure IAM ---
+			externalUrl := tt.externalUrl
+			clientHost := tt.clientHost
+			clientScheme := tt.clientScheme
+			if externalUrl == "" && clientHost == "" {
+				// Dynamic: use the proxy's actual address
+				proxyURL, _ := url.Parse(proxyServer.URL)
+				externalUrl = proxyServer.URL
+				clientHost = proxyURL.Host
+				clientScheme = proxyURL.Scheme
+			}
+
+			option := &S3ApiServerOption{
+				Config:      tmpFile,
+				ExternalUrl: externalUrl,
+			}
+			iam = NewIdentityAccessManagementWithStore(option, nil, "memory")
+			require.True(t, iam.isEnabled())
+
+			// --- Sign the request using real AWS SDK v4 signer ---
+			clientURL := fmt.Sprintf("%s://%s/test-bucket/test-object", clientScheme, clientHost)
+			req, err := http.NewRequest(http.MethodGet, clientURL, nil)
+			require.NoError(t, err)
+			req.Host = clientHost
+
+			signer := v4.NewSigner()
+			payloadHash := fmt.Sprintf("%x", sha256.Sum256([]byte{}))
+			err = signer.SignHTTP(
+				context.Background(),
+				aws.Credentials{AccessKeyID: accessKey, SecretAccessKey: secretKey},
+				req, payloadHash, "s3", "us-east-1", time.Now(),
+			)
+			require.NoError(t, err)
+
+			// --- Send the signed request through the proxy ---
+			// Rewrite destination to the proxy, but keep signed headers and Host intact
+			proxyURL, _ := url.Parse(proxyServer.URL)
+			req.URL.Scheme = proxyURL.Scheme
+			req.URL.Host = proxyURL.Host
+
+			resp, err := http.DefaultClient.Do(req)
+			require.NoError(t, err)
+			defer resp.Body.Close()
+
+			if tt.expectSuccess {
+				assert.Equal(t, http.StatusOK, resp.StatusCode,
+					"Expected signature verification to succeed through reverse proxy")
+			} else {
+				assert.Equal(t, http.StatusForbidden, resp.StatusCode,
+					"Expected signature verification to fail (host mismatch)")
+			}
+		})
+	}
+}

weed/s3api/auth_security_test.go

@@ -179,3 +179,298 @@ func TestReproIssue7912(t *testing.T) {
 		assert.Nil(t, identity)
 	})
 }
+
+// TestExternalUrlSignatureVerification tests that S3 signature verification works
+// correctly when s3.externalUrl is configured. It uses the real AWS SDK v2 signer
+// to prove correctness against actual S3 clients behind a reverse proxy.
+func TestExternalUrlSignatureVerification(t *testing.T) {
+	configContent := `{
+  "identities": [
+    {
+      "name": "test_user",
+      "credentials": [
+        {
+          "accessKey": "AKIAIOSFODNN7EXAMPLE",
+          "secretKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+        }
+      ],
+      "actions": ["Admin", "Read", "Write", "List", "Tagging"]
+    }
+  ]
+}`
+	tmpFile, err := os.CreateTemp("", "s3-config-*.json")
+	require.NoError(t, err)
+	defer os.Remove(tmpFile.Name())
+	_, err = tmpFile.Write([]byte(configContent))
+	require.NoError(t, err)
+	tmpFile.Close()
+
+	accessKey := "AKIAIOSFODNN7EXAMPLE"
+	secretKey := "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+
+	tests := []struct {
+		name          string
+		clientUrl     string // URL the client signs against
+		backendHost   string // Host header SeaweedFS sees (from proxy)
+		externalUrl   string // s3.externalUrl config
+		expectSuccess bool
+	}{
+		{
+			name:          "non-standard port with externalUrl",
+			clientUrl:     "https://api.example.com:9000/test-bucket/object",
+			backendHost:   "backend:8333",
+			externalUrl:   "https://api.example.com:9000",
+			expectSuccess: true,
+		},
+		{
+			name:          "HTTPS default port 443 with externalUrl (port stripped by SDK and parseExternalUrlToHost)",
+			clientUrl:     "https://api.example.com/test-bucket/object",
+			backendHost:   "backend:8333",
+			externalUrl:   "https://api.example.com:443",
+			expectSuccess: true,
+		},
+		{
+			name:          "HTTPS without explicit port with externalUrl",
+			clientUrl:     "https://api.example.com/test-bucket/object",
+			backendHost:   "backend:8333",
+			externalUrl:   "https://api.example.com",
+			expectSuccess: true,
+		},
+		{
+			name:          "HTTP default port 80 with externalUrl",
+			clientUrl:     "http://api.example.com/test-bucket/object",
+			backendHost:   "backend:8333",
+			externalUrl:   "http://api.example.com:80",
+			expectSuccess: true,
+		},
+		{
+			name:          "without externalUrl, internal host causes mismatch",
+			clientUrl:     "https://api.example.com:9000/test-bucket/object",
+			backendHost:   "backend:8333",
+			externalUrl:   "",
+			expectSuccess: false,
+		},
+		{
+			name:          "without externalUrl, matching host works",
+			clientUrl:     "http://localhost:8333/test-bucket/object",
+			backendHost:   "localhost:8333",
+			externalUrl:   "",
+			expectSuccess: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create IAM with the externalUrl
+			option := &S3ApiServerOption{
+				Config:      tmpFile.Name(),
+				ExternalUrl: tt.externalUrl,
+			}
+			iam := NewIdentityAccessManagementWithStore(option, nil, "memory")
+			require.True(t, iam.isEnabled())
+
+			// Step 1: Sign a request targeting the client-facing URL using real AWS SDK signer
+			signReq, err := http.NewRequest(http.MethodGet, tt.clientUrl, nil)
+			require.NoError(t, err)
+			signReq.Host = signReq.URL.Host
+
+			err = signRawHTTPRequest(context.Background(), signReq, accessKey, secretKey, "us-east-1")
+			require.NoError(t, err)
+
+			// Step 2: Create a separate request as the proxy would deliver it
+			proxyReq := httptest.NewRequest(http.MethodGet, tt.clientUrl, nil)
+			proxyReq.Host = tt.backendHost
+			proxyReq.URL.Host = tt.backendHost
+
+			// Copy the auth headers from the signed request
+			proxyReq.Header.Set("Authorization", signReq.Header.Get("Authorization"))
+			proxyReq.Header.Set("X-Amz-Date", signReq.Header.Get("X-Amz-Date"))
+			proxyReq.Header.Set("X-Amz-Content-Sha256", signReq.Header.Get("X-Amz-Content-Sha256"))
+
+			// Step 3: Verify
+			identity, errCode := iam.authRequest(proxyReq, s3_constants.ACTION_LIST)
+			if tt.expectSuccess {
+				assert.Equal(t, s3err.ErrNone, errCode, "Expected successful signature verification")
+				require.NotNil(t, identity)
+				assert.Equal(t, "test_user", identity.Name)
+			} else {
+				assert.NotEqual(t, s3err.ErrNone, errCode, "Expected signature verification to fail")
+			}
+		})
+	}
+}
+
+// TestRealSDKSignerWithForwardedHeaders proves that extractHostHeader produces
+// values that match the real AWS SDK v2 signer's SanitizeHostForHeader behavior.
+//
+// Unlike the self-referential tests in auto_signature_v4_test.go (which sign and
+// verify with the same extractHostHeader function), this test uses the real AWS
+// SDK v2 signer to sign the request. The SDK has its own host sanitization:
+//   - strips :80 for http
+//   - strips :443 for https
+//   - preserves all other ports
+//
+// If extractHostHeader disagrees with the SDK, the signature will not match.
+func TestRealSDKSignerWithForwardedHeaders(t *testing.T) {
+	configContent := `{
+  "identities": [
+    {
+      "name": "test_user",
+      "credentials": [
+        {
+          "accessKey": "AKIAIOSFODNN7EXAMPLE",
+          "secretKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+        }
+      ],
+      "actions": ["Admin", "Read", "Write", "List", "Tagging"]
+    }
+  ]
+}`
+	tmpFile, err := os.CreateTemp("", "s3-config-*.json")
+	require.NoError(t, err)
+	defer os.Remove(tmpFile.Name())
+	_, err = tmpFile.Write([]byte(configContent))
+	require.NoError(t, err)
+	tmpFile.Close()
+
+	accessKey := "AKIAIOSFODNN7EXAMPLE"
+	secretKey := "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+
+	// Each test case simulates:
+	//   1. A client connecting to clientUrl (SDK signs with the host from this URL)
+	//   2. A proxy forwarding to SeaweedFS with X-Forwarded-* headers
+	//   3. SeaweedFS extracting the host and verifying the signature
+	tests := []struct {
+		name           string
+		clientUrl      string // URL the client/SDK signs against
+		backendHost    string // Host header SeaweedFS receives (proxy rewrites this)
+		forwardedHost  string // X-Forwarded-Host set by proxy
+		forwardedPort  string // X-Forwarded-Port set by proxy
+		forwardedProto string // X-Forwarded-Proto set by proxy
+		expectedHost   string // what extractHostHeader should return (must match SDK)
+	}{
+		{
+			name:           "HTTPS non-standard port",
+			clientUrl:      "https://api.example.com:9000/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com",
+			forwardedPort:  "9000",
+			forwardedProto: "https",
+			expectedHost:   "api.example.com:9000",
+		},
+		{
+			name:           "HTTPS standard port 443 (SDK strips, we must too)",
+			clientUrl:      "https://api.example.com/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com",
+			forwardedPort:  "443",
+			forwardedProto: "https",
+			expectedHost:   "api.example.com",
+		},
+		{
+			name:           "HTTP standard port 80 (SDK strips, we must too)",
+			clientUrl:      "http://api.example.com/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com",
+			forwardedPort:  "80",
+			forwardedProto: "http",
+			expectedHost:   "api.example.com",
+		},
+		{
+			name:           "HTTP non-standard port 8080",
+			clientUrl:      "http://api.example.com:8080/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com",
+			forwardedPort:  "8080",
+			forwardedProto: "http",
+			expectedHost:   "api.example.com:8080",
+		},
+		{
+			name:           "X-Forwarded-Host with port (Traefik style), HTTPS 443",
+			clientUrl:      "https://api.example.com/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com:443",
+			forwardedPort:  "443",
+			forwardedProto: "https",
+			expectedHost:   "api.example.com",
+		},
+		{
+			name:           "X-Forwarded-Host with port (Traefik style), HTTP 80",
+			clientUrl:      "http://api.example.com/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com:80",
+			forwardedPort:  "80",
+			forwardedProto: "http",
+			expectedHost:   "api.example.com",
+		},
+		{
+			name:           "no forwarded headers, direct access",
+			clientUrl:      "http://localhost:8333/bucket/key",
+			backendHost:    "localhost:8333",
+			forwardedHost:  "",
+			forwardedPort:  "",
+			forwardedProto: "",
+			expectedHost:   "localhost:8333",
+		},
+		{
+			name:           "empty proto with port 80 (defaults to http, strip)",
+			clientUrl:      "http://api.example.com/bucket/key",
+			backendHost:    "seaweedfs:8333",
+			forwardedHost:  "api.example.com",
+			forwardedPort:  "80",
+			forwardedProto: "",
+			expectedHost:   "api.example.com",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			option := &S3ApiServerOption{Config: tmpFile.Name()}
+			iam := NewIdentityAccessManagementWithStore(option, nil, "memory")
+			require.True(t, iam.isEnabled())
+
+			// Step 1: Sign a request using the real AWS SDK v2 signer.
+			// The SDK applies its own SanitizeHostForHeader logic.
+			signReq, err := http.NewRequest(http.MethodGet, tt.clientUrl, nil)
+			require.NoError(t, err)
+			signReq.Host = signReq.URL.Host
+
+			err = signRawHTTPRequest(context.Background(), signReq, accessKey, secretKey, "us-east-1")
+			require.NoError(t, err)
+
+			// Step 2: Build the request as the proxy would deliver it to SeaweedFS.
+			proxyReq := httptest.NewRequest(http.MethodGet, tt.clientUrl, nil)
+			proxyReq.Host = tt.backendHost
+
+			// Copy signed auth headers
+			proxyReq.Header.Set("Authorization", signReq.Header.Get("Authorization"))
+			proxyReq.Header.Set("X-Amz-Date", signReq.Header.Get("X-Amz-Date"))
+			proxyReq.Header.Set("X-Amz-Content-Sha256", signReq.Header.Get("X-Amz-Content-Sha256"))
+
+			// Set forwarded headers
+			if tt.forwardedHost != "" {
+				proxyReq.Header.Set("X-Forwarded-Host", tt.forwardedHost)
+			}
+			if tt.forwardedPort != "" {
+				proxyReq.Header.Set("X-Forwarded-Port", tt.forwardedPort)
+			}
+			if tt.forwardedProto != "" {
+				proxyReq.Header.Set("X-Forwarded-Proto", tt.forwardedProto)
+			}
+
+			// Step 3: Verify extractHostHeader returns the expected value.
+			// This is the critical correctness check: our extracted host must match
+			// what the SDK signed with, otherwise the signature will not match.
+			extractedHost := extractHostHeader(proxyReq, iam.externalHost)
+			assert.Equal(t, tt.expectedHost, extractedHost,
+				"extractHostHeader must return a value matching what AWS SDK signed with")
+
+			// Step 4: Verify the full signature matches.
+			identity, errCode := iam.authRequest(proxyReq, s3_constants.ACTION_LIST)
+			assert.Equal(t, s3err.ErrNone, errCode,
+				"Signature from real AWS SDK signer must verify successfully")
+			require.NotNil(t, identity)
+			assert.Equal(t, "test_user", identity.Name)
+		})
+	}
+}

weed/s3api/auth_signature_v4.go

@@ -284,7 +284,7 @@ func (iam *IdentityAccessManagement) verifyV4Signature(r *http.Request, shouldCh
 	}
 
 	// 5. Extract headers that were part of the signature
-	extractedSignedHeaders, errCode := extractSignedHeaders(authInfo.SignedHeaders, r)
+	extractedSignedHeaders, errCode := extractSignedHeaders(authInfo.SignedHeaders, r, iam.externalHost)
 	if errCode != s3err.ErrNone {
 		return nil, nil, "", nil, errCode
 	}
@@ -760,7 +760,7 @@ func (iam *IdentityAccessManagement) doesPolicySignatureV4Match(formValues http.
 }
 
 // Verify if extracted signed headers are not properly signed.
-func extractSignedHeaders(signedHeaders []string, r *http.Request) (http.Header, s3err.ErrorCode) {
+func extractSignedHeaders(signedHeaders []string, r *http.Request, externalHost string) (http.Header, s3err.ErrorCode) {
 	reqHeaders := r.Header
 	// If no signed headers are provided, then return an error.
 	if len(signedHeaders) == 0 {
@@ -771,7 +771,7 @@ func extractSignedHeaders(signedHeaders []string, r *http.Request) (http.Header,
 		// `host` is not a case-sensitive header, unlike other headers such as `x-amz-date`.
 		if strings.ToLower(header) == "host" {
 			// Get host value.
-			hostHeaderValue := extractHostHeader(r)
+			hostHeaderValue := extractHostHeader(r, externalHost)
 			extractedSignedHeaders[header] = []string{hostHeaderValue}
 			continue
 		}
@@ -784,17 +784,30 @@ func extractSignedHeaders(signedHeaders []string, r *http.Request) (http.Header,
 	return extractedSignedHeaders, s3err.ErrNone
 }
 
-// extractHostHeader returns the value of host header if available.
-func extractHostHeader(r *http.Request) string {
+// extractHostHeader returns the value of host header to use for signature verification.
+// When externalHost is set (from s3.externalUrl), it is returned directly.
+// Otherwise, the host is reconstructed from X-Forwarded-* headers or the request Host,
+// with default port stripping to match AWS SDK SanitizeHostForHeader behavior.
+func extractHostHeader(r *http.Request, externalHost string) string {
+	if externalHost != "" {
+		return externalHost
+	}
+
 	forwardedHost := r.Header.Get("X-Forwarded-Host")
 	forwardedPort := r.Header.Get("X-Forwarded-Port")
 	forwardedProto := r.Header.Get("X-Forwarded-Proto")
 
-	// Determine the effective scheme with correct order of precedence:
-	// 1. X-Forwarded-Proto (most authoritative, reflects client's original protocol)
-	// 2. r.TLS (authoritative for direct connection to server)
-	// 3. r.URL.Scheme (fallback, may not always be set correctly)
-	// 4. Default to "http"
+	// X-Forwarded-Proto and X-Forwarded-Port can be comma-separated lists when there are multiple proxies.
+	// Use only the first value (first-hop).
+	if comma := strings.Index(forwardedPort, ","); comma != -1 {
+		forwardedPort = strings.TrimSpace(forwardedPort[:comma])
+	}
+	if comma := strings.Index(forwardedProto, ","); comma != -1 {
+		forwardedProto = strings.TrimSpace(forwardedProto[:comma])
+	}
+
+	// Determine effective scheme for default port stripping.
+	// Precedence: X-Forwarded-Proto > r.TLS > r.URL.Scheme > "http"
 	scheme := "http"
 	if r.URL.Scheme != "" {
 		scheme = r.URL.Scheme
@@ -815,51 +828,55 @@ func extractHostHeader(r *http.Request) string {
 		} else {
 			host = strings.TrimSpace(forwardedHost)
 		}
+
+		// If the host itself contains a port, it should take precedence
 		if h, p, err := net.SplitHostPort(host); err == nil {
 			host = h
 			port = p
-		}
-		if forwardedPort != "" && isDefaultPort(scheme, port) {
-			port = forwardedPort
+		} else {
+			// If X-Forwarded-Host has no port, try to get port from r.Host if hostnames match
+			if rh, rp, err := net.SplitHostPort(r.Host); err == nil && rh == host {
+				port = rp
+			} else if forwardedPort != "" {
+				port = forwardedPort
+			}
 		}
 	} else {
 		host = r.Host
 		if host == "" {
 			host = r.URL.Host
 		}
+
+		// If the host already contains a port, use it.
+		// Otherwise, if X-Forwarded-Port is set, use it.
 		if h, p, err := net.SplitHostPort(host); err == nil {
 			host = h
 			port = p
+		} else if forwardedPort != "" {
+			port = forwardedPort
 		}
 	}
 
-	// If we have a non-default port, join it with the host.
-	// net.JoinHostPort will handle bracketing for IPv6.
+	// Strip default ports based on scheme to match AWS SDK SanitizeHostForHeader behavior.
+	// The AWS SDK strips port 80 for HTTP and port 443 for HTTPS before signing.
 	if port != "" && !isDefaultPort(scheme, port) {
 		// Strip existing brackets before calling JoinHostPort, which automatically adds
 		// brackets for IPv6 addresses. This prevents double-bracketing like [[::1]]:8080.
-		// Using Trim handles both well-formed and malformed bracketed hosts.
 		host = strings.Trim(host, "[]")
 		return net.JoinHostPort(host, port)
 	}
 
-	// No port or default port was stripped. According to AWS SDK behavior (aws-sdk-go-v2),
-	// when a default port is removed from an IPv6 address, the brackets should also be removed.
-	// This matches AWS S3 signature calculation requirements.
+	// Default port was stripped, or no port present.
+	// For IPv6 addresses, strip brackets to match AWS SDK behavior.
 	// Reference: https://github.com/aws/aws-sdk-go-v2/blob/main/aws/signer/internal/v4/host.go
-	// The stripPort function returns IPv6 without brackets when port is stripped.
 	if strings.Contains(host, ":") {
-		// This is an IPv6 address. Strip brackets to match AWS SDK behavior.
 		return strings.Trim(host, "[]")
 	}
 	return host
 }
 
+// isDefaultPort returns true if the given port is the default for the scheme.
 func isDefaultPort(scheme, port string) bool {
-	if port == "" {
-		return true
-	}
-
 	switch port {
 	case "80":
 		return strings.EqualFold(scheme, "http")

weed/s3api/auth_signature_v4_test.go

@@ -160,6 +160,7 @@ func TestExtractHostHeader(t *testing.T) {
 		forwardedHost  string
 		forwardedPort  string
 		forwardedProto string
+		externalHost   string
 		expected       string
 	}{
 		{
@@ -227,6 +228,14 @@ func TestExtractHostHeader(t *testing.T) {
 			forwardedProto: "https",
 			expected:       "127.0.0.1:8433",
 		},
+		{
+			name:           "X-Forwarded-Host with standard port already included (HTTPS 443)",
+			hostHeader:     "backend:8333",
+			forwardedHost:  "example.com:443",
+			forwardedPort:  "443",
+			forwardedProto: "https",
+			expected:       "example.com",
+		},
 		{
 			name:           "X-Forwarded-Host with port, no X-Forwarded-Port header",
 			hostHeader:     "backend:8333",
@@ -253,7 +262,7 @@ func TestExtractHostHeader(t *testing.T) {
 			expected:       "[::1]:8080",
 		},
 		{
-			name:           "IPv6 address without brackets and standard port, should strip brackets per AWS SDK",
+			name:           "IPv6 address without brackets and standard port, should strip default port",
 			hostHeader:     "backend:8333",
 			forwardedHost:  "::1",
 			forwardedPort:  "80",
@@ -261,7 +270,7 @@ func TestExtractHostHeader(t *testing.T) {
 			expected:       "::1",
 		},
 		{
-			name:           "IPv6 address without brackets and standard HTTPS port, should strip brackets per AWS SDK",
+			name:           "IPv6 address without brackets and standard HTTPS port, should strip default port",
 			hostHeader:     "backend:8333",
 			forwardedHost:  "2001:db8::1",
 			forwardedPort:  "443",
@@ -277,7 +286,7 @@ func TestExtractHostHeader(t *testing.T) {
 			expected:       "[2001:db8::1]:8080",
 		},
 		{
-			name:           "IPv6 full address with brackets and default port (should strip port and brackets)",
+			name:           "IPv6 full address with brackets and default port (should strip default port)",
 			hostHeader:     "backend:8333",
 			forwardedHost:  "[2001:db8:85a3::8a2e:370:7334]:443",
 			forwardedPort:  "443",
@@ -333,6 +342,53 @@ func TestExtractHostHeader(t *testing.T) {
 			forwardedProto: "http",
 			expected:       "bucket.domain.com:442",
 		},
+		// externalHost override tests
+		{
+			name:         "externalHost overrides everything",
+			hostHeader:   "backend:8333",
+			externalHost: "api.example.com:9000",
+			expected:     "api.example.com:9000",
+		},
+		{
+			name:           "externalHost overrides X-Forwarded-Host",
+			hostHeader:     "backend:8333",
+			forwardedHost:  "proxy.example.com",
+			forwardedPort:  "443",
+			forwardedProto: "https",
+			externalHost:   "api.example.com",
+			expected:       "api.example.com",
+		},
+		{
+			name:         "externalHost with IPv6",
+			hostHeader:   "backend:8333",
+			externalHost: "[::1]:9000",
+			expected:     "[::1]:9000",
+		},
+		// Bug fix: X-Forwarded-Port should not override more specific ports in other headers
+		{
+			name:           "User reported case: X-Forwarded-Port misreports 443 but Host has 30007",
+			hostHeader:     "storage-stgops.mt.mtnet:30007",
+			forwardedHost:  "storage-stgops.mt.mtnet",
+			forwardedPort:  "443",
+			forwardedProto: "https",
+			expected:       "storage-stgops.mt.mtnet:30007",
+		},
+		{
+			name:           "X-Forwarded-Host already contains correct port, ignore misaligned X-Forwarded-Port",
+			hostHeader:     "backend:8333",
+			forwardedHost:  "storage-stgops.mt.mtnet:30007",
+			forwardedPort:  "443",
+			forwardedProto: "https",
+			expected:       "storage-stgops.mt.mtnet:30007",
+		},
+		{
+			name:           "X-Forwarded-Host has no port, match r.Host hostname and take its port",
+			hostHeader:     "example.com:8080",
+			forwardedHost:  "example.com",
+			forwardedPort:  "80",
+			forwardedProto: "http",
+			expected:       "example.com:8080",
+		},
 	}
 
 	for _, tt := range tests {
@@ -356,7 +412,7 @@ func TestExtractHostHeader(t *testing.T) {
 			}
 
 			// Test the function
-			result := extractHostHeader(req)
+			result := extractHostHeader(req, tt.externalHost)
 			if result != tt.expected {
 				t.Errorf("extractHostHeader() = %q, want %q", result, tt.expected)
 			}
@@ -389,7 +445,7 @@ func TestExtractSignedHeadersCase(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			r, _ := http.NewRequest("GET", "http://"+tt.host+"/", nil)
 			r.Host = tt.host
-			extracted, errCode := extractSignedHeaders(tt.signedHeads, r)
+			extracted, errCode := extractSignedHeaders(tt.signedHeads, r, "")
 			if errCode != s3err.ErrNone {
 				t.Fatalf("extractSignedHeaders failed: %v", errCode)
 			}

weed/s3api/auto_signature_v4_test.go

@@ -493,7 +493,7 @@ func TestSignatureV4WithoutProxy(t *testing.T) {
 			r.Header.Set("Host", tt.host)
 
 			// First, verify that extractHostHeader returns the expected value
-			extractedHost := extractHostHeader(r)
+			extractedHost := extractHostHeader(r, "")
 			if extractedHost != tt.expectedHost {
 				t.Errorf("extractHostHeader() = %q, want %q", extractedHost, tt.expectedHost)
 			}
@@ -562,12 +562,12 @@ func TestSignatureV4WithForwardedPort(t *testing.T) {
 			expectedHost:   "example.com:8080",
 		},
 		{
-			name:           "empty proto with standard http port",
+			name:           "empty proto with port 80 (scheme defaults to https from URL, so 80 is NOT default)",
 			host:           "backend:8333",
 			forwardedHost:  "example.com",
 			forwardedPort:  "80",
 			forwardedProto: "",
-			expectedHost:   "example.com",
+			expectedHost:   "example.com:80",
 		},
 		// Test cases for issue #6649: X-Forwarded-Host already contains port
 		{
@@ -674,8 +674,16 @@ func TestSignatureV4WithForwardedPort(t *testing.T) {
 			r.Header.Set("X-Forwarded-Port", tt.forwardedPort)
 			r.Header.Set("X-Forwarded-Proto", tt.forwardedProto)
 
-			// Sign the request with the expected host header
-			// We need to temporarily modify the Host header for signing
+			// Validate that extractHostHeader returns the expected host value.
+			// This is critical: the expectedHost must match what the AWS SDK would
+			// use for signing. Without this check, the test is self-referential
+			// (signing and verifying with the same function always agrees).
+			extractedHost := extractHostHeader(r, "")
+			if extractedHost != tt.expectedHost {
+				t.Errorf("extractHostHeader() = %q, want %q", extractedHost, tt.expectedHost)
+			}
+
+			// Sign the request (note: signV4WithPath uses extractHostHeader internally)
 			signV4WithPath(r, "AKIAIOSFODNN7EXAMPLE", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", r.URL.Path)
 
 			// Test signature verification
@@ -922,7 +930,7 @@ func preSignV4WithPath(iam *IdentityAccessManagement, req *http.Request, accessK
 
 	// Extract signed headers
 	extractedSignedHeaders := make(http.Header)
-	extractedSignedHeaders["host"] = []string{extractHostHeader(req)}
+	extractedSignedHeaders["host"] = []string{extractHostHeader(req, "")}
 
 	// Get canonical request with custom path
 	canonicalRequest := getCanonicalRequest(extractedSignedHeaders, hashedPayload, req.URL.RawQuery, urlPath, req.Method)
@@ -961,7 +969,7 @@ func signV4WithPath(req *http.Request, accessKey, secretKey, urlPath string) {
 
 	// Extract signed headers
 	extractedSignedHeaders := make(http.Header)
-	extractedSignedHeaders["host"] = []string{extractHostHeader(req)}
+	extractedSignedHeaders["host"] = []string{extractHostHeader(req, "")}
 	extractedSignedHeaders["x-amz-date"] = []string{dateStr}
 
 	// Get the payload hash

weed/s3api/bucket_metadata.go

@@ -49,9 +49,6 @@ type BucketRegistry struct {
 	metadataCache     map[string]*BucketMetaData
 	metadataCacheLock sync.RWMutex
 
-	tableLocationCache map[string]string // Cache for table location mappings (external bucket -> mapped root path)
-	tableLocationLock  sync.RWMutex
-
 	notFound     map[string]struct{}
 	notFoundLock sync.RWMutex
 	s3a          *S3ApiServer
@@ -59,10 +56,9 @@ type BucketRegistry struct {
 
 func NewBucketRegistry(s3a *S3ApiServer) *BucketRegistry {
 	br := &BucketRegistry{
-		metadataCache:      make(map[string]*BucketMetaData),
-		tableLocationCache: make(map[string]string),
-		notFound:           make(map[string]struct{}),
-		s3a:                s3a,
+		metadataCache: make(map[string]*BucketMetaData),
+		notFound:      make(map[string]struct{}),
+		s3a:           s3a,
 	}
 	err := br.init()
 	if err != nil {
@@ -164,7 +160,6 @@ func buildBucketMetadata(accountManager AccountManager, entry *filer_pb.Entry) *
 func (r *BucketRegistry) RemoveBucketMetadata(entry *filer_pb.Entry) {
 	r.removeMetadataCache(entry.Name)
 	r.unMarkNotFound(entry.Name)
-	r.removeTableLocationCache(entry.Name)
 }
 
 func (r *BucketRegistry) GetBucketMetadata(bucketName string) (*BucketMetaData, s3err.ErrorCode) {
@@ -229,12 +224,6 @@ func (r *BucketRegistry) removeMetadataCache(bucket string) {
 	delete(r.metadataCache, bucket)
 }
 
-func (r *BucketRegistry) removeTableLocationCache(bucket string) {
-	r.tableLocationLock.Lock()
-	defer r.tableLocationLock.Unlock()
-	delete(r.tableLocationCache, bucket)
-}
-
 func (r *BucketRegistry) markNotFound(bucket string) {
 	r.notFoundLock.Lock()
 	defer r.notFoundLock.Unlock()

weed/s3api/bucket_paths.go

@@ -1,9 +1,7 @@
 package s3api
 
 import (
-	"context"
 	"errors"
-	"io"
 	"path"
 	"strings"
 
@@ -43,132 +41,12 @@ func (s3a *S3ApiServer) isTableBucket(bucket string) bool {
 	return false
 }
 
-func (s3a *S3ApiServer) tableLocationDir(bucket string) (string, bool) {
-	if bucket == "" {
-		return "", false
-	}
-
-	// Check cache first
-	if s3a.bucketRegistry != nil {
-		s3a.bucketRegistry.tableLocationLock.RLock()
-		if tablePath, ok := s3a.bucketRegistry.tableLocationCache[bucket]; ok {
-			s3a.bucketRegistry.tableLocationLock.RUnlock()
-			return tablePath, tablePath != ""
-		}
-		s3a.bucketRegistry.tableLocationLock.RUnlock()
-	}
-
-	entry, err := s3a.getEntry(s3tables.GetTableLocationMappingDir(), bucket)
-	tablePath := ""
-	if err == nil && entry != nil {
-		if entry.IsDirectory {
-			tablePath, err = s3a.readTableLocationMappingFromDirectory(bucket)
-		} else if len(entry.Content) > 0 {
-			// Backward compatibility with legacy single-file mappings.
-			tablePath = normalizeTableLocationMappingPath(string(entry.Content))
-		}
-	}
-
-	// Only cache definitive results: successful lookup (tablePath set) or definitive not-found (ErrNotFound)
-	// Don't cache transient errors to avoid treating temporary failures as permanent misses
-	if err == nil || errors.Is(err, filer_pb.ErrNotFound) {
-		if s3a.bucketRegistry != nil {
-			s3a.bucketRegistry.tableLocationLock.Lock()
-			s3a.bucketRegistry.tableLocationCache[bucket] = tablePath
-			s3a.bucketRegistry.tableLocationLock.Unlock()
-		}
-	}
-
-	if tablePath == "" {
-		if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
-			glog.V(1).Infof("table location mapping lookup failed for %s: %v", bucket, err)
-		}
-		return "", false
-	}
-
-	return tablePath, true
-}
-
-func (s3a *S3ApiServer) readTableLocationMappingFromDirectory(bucket string) (string, error) {
-	mappingDir := s3tables.GetTableLocationMappingPath(bucket)
-	var mappedPath string
-	conflict := false
-
-	err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
-			Directory: mappingDir,
-			Limit:     4294967295, // math.MaxUint32
-		})
-		if err != nil {
-			return err
-		}
-
-		for {
-			resp, recvErr := stream.Recv()
-			if recvErr == io.EOF {
-				return nil
-			}
-			if recvErr != nil {
-				return recvErr
-			}
-			if resp == nil || resp.Entry == nil || resp.Entry.IsDirectory || len(resp.Entry.Content) == 0 {
-				continue
-			}
-
-			candidate := normalizeTableLocationMappingPath(string(resp.Entry.Content))
-			if candidate == "" {
-				continue
-			}
-			if mappedPath == "" {
-				mappedPath = candidate
-				continue
-			}
-			if mappedPath != candidate {
-				conflict = true
-				return nil
-			}
-		}
-	})
-	if err != nil {
-		return "", err
-	}
-
-	if conflict {
-		glog.V(1).Infof("table location mapping conflict for %s: multiple mapped roots found", bucket)
-		return "", nil
-	}
-	return mappedPath, nil
-}
-
-func normalizeTableLocationMappingPath(rawPath string) string {
-	rawPath = strings.TrimSpace(rawPath)
-	if rawPath == "" {
-		return ""
-	}
-
-	normalized := path.Clean("/" + strings.TrimPrefix(rawPath, "/"))
-	tablesPrefix := strings.TrimSuffix(s3tables.TablesPath, "/") + "/"
-	if !strings.HasPrefix(normalized, tablesPrefix) {
-		return normalized
-	}
-
-	remaining := strings.TrimPrefix(normalized, tablesPrefix)
-	bucketName, _, _ := strings.Cut(remaining, "/")
-	if bucketName == "" {
-		return ""
-	}
-	return path.Join(s3tables.TablesPath, bucketName)
-}
-
 func (s3a *S3ApiServer) bucketRoot(bucket string) string {
 	// Returns the unified buckets root path for all bucket types
 	return s3a.option.BucketsPath
 }
 
 func (s3a *S3ApiServer) bucketDir(bucket string) string {
-	if tablePath, ok := s3a.tableLocationDir(bucket); ok {
-		return tablePath
-	}
 	return path.Join(s3a.bucketRoot(bucket), bucket)
 }
 
@@ -217,8 +95,5 @@ func (s3a *S3ApiServer) bucketExists(bucket string) (bool, error) {
 }
 
 func (s3a *S3ApiServer) getBucketEntry(bucket string) (*filer_pb.Entry, error) {
-	if tablePath, ok := s3a.tableLocationDir(bucket); ok {
-		return s3a.getEntry(path.Dir(tablePath), path.Base(tablePath))
-	}
 	return s3a.getEntry(s3a.option.BucketsPath, bucket)
 }

weed/s3api/bucket_paths_test.go

@@ -1,45 +0,0 @@
-package s3api
-
-import "testing"
-
-func TestNormalizeTableLocationMappingPath(t *testing.T) {
-	testCases := []struct {
-		name string
-		raw  string
-		want string
-	}{
-		{
-			name: "legacy table path maps to bucket root",
-			raw:  "/buckets/warehouse/analytics/orders",
-			want: "/buckets/warehouse",
-		},
-		{
-			name: "already bucket root",
-			raw:  "/buckets/warehouse",
-			want: "/buckets/warehouse",
-		},
-		{
-			name: "relative buckets path normalized and reduced",
-			raw:  "buckets/warehouse/analytics/orders",
-			want: "/buckets/warehouse",
-		},
-		{
-			name: "non buckets path preserved",
-			raw:  "/tmp/custom/path",
-			want: "/tmp/custom/path",
-		},
-		{
-			name: "empty path",
-			raw:  "",
-			want: "",
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			if got := normalizeTableLocationMappingPath(tc.raw); got != tc.want {
-				t.Fatalf("normalizeTableLocationMappingPath(%q)=%q, want %q", tc.raw, got, tc.want)
-			}
-		})
-	}
-}

weed/s3api/policy_engine/conditions.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // LRUNode represents a node in the doubly-linked list for efficient LRU operations
@@ -210,7 +211,7 @@ func (e *StringLikeEvaluator) Evaluate(conditionValue interface{}, contextValues
 	patterns := getCachedNormalizedValues(conditionValue)
 	for _, pattern := range patterns {
 		for _, contextValue := range contextValues {
-			if MatchesWildcard(pattern, contextValue) {
+			if wildcard.MatchesWildcard(pattern, contextValue) {
 				return true
 			}
 		}
@@ -225,7 +226,7 @@ func (e *StringNotLikeEvaluator) Evaluate(conditionValue interface{}, contextVal
 	patterns := getCachedNormalizedValues(conditionValue)
 	for _, pattern := range patterns {
 		for _, contextValue := range contextValues {
-			if MatchesWildcard(pattern, contextValue) {
+			if wildcard.MatchesWildcard(pattern, contextValue) {
 				return false
 			}
 		}
@@ -628,7 +629,7 @@ func (e *ArnLikeEvaluator) Evaluate(conditionValue interface{}, contextValues []
 	patterns := getCachedNormalizedValues(conditionValue)
 	for _, pattern := range patterns {
 		for _, contextValue := range contextValues {
-			if MatchesWildcard(pattern, contextValue) {
+			if wildcard.MatchesWildcard(pattern, contextValue) {
 				return true
 			}
 		}

weed/s3api/policy_engine/engine_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // tagsToEntry converts a map of tag key-value pairs to the entry.Extended format
@@ -749,7 +750,7 @@ func TestWildcardMatching(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := MatchesWildcard(tt.pattern, tt.str)
+			result := wildcard.MatchesWildcard(tt.pattern, tt.str)
 			if result != tt.expected {
 				t.Errorf("Pattern %s against %s: expected %v, got %v", tt.pattern, tt.str, tt.expected, result)
 			}

weed/s3api/policy_engine/types.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	s3const "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // Policy Engine Types
@@ -180,9 +181,9 @@ type CompiledPolicy struct {
 // CompiledStatement represents a compiled policy statement
 type CompiledStatement struct {
 	Statement         *PolicyStatement
-	ActionMatchers    []*WildcardMatcher
-	ResourceMatchers  []*WildcardMatcher
-	PrincipalMatchers []*WildcardMatcher
+	ActionMatchers    []*wildcard.WildcardMatcher
+	ResourceMatchers  []*wildcard.WildcardMatcher
+	PrincipalMatchers []*wildcard.WildcardMatcher
 	// Keep regex patterns for backward compatibility
 	ActionPatterns    []*regexp.Regexp
 	ResourcePatterns  []*regexp.Regexp
@@ -195,7 +196,7 @@ type CompiledStatement struct {
 
 	// NotResource patterns (resource should NOT match these)
 	NotResourcePatterns        []*regexp.Regexp
-	NotResourceMatchers        []*WildcardMatcher
+	NotResourceMatchers        []*wildcard.WildcardMatcher
 	DynamicNotResourcePatterns []string
 }
 
@@ -328,7 +329,7 @@ func compileStatement(stmt *PolicyStatement) (*CompiledStatement, error) {
 		}
 		compiled.ActionPatterns = append(compiled.ActionPatterns, pattern)
 
-		matcher, err := NewWildcardMatcher(action)
+		matcher, err := wildcard.NewWildcardMatcher(action)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create action matcher %s: %v", action, err)
 		}
@@ -352,7 +353,7 @@ func compileStatement(stmt *PolicyStatement) (*CompiledStatement, error) {
 		}
 		compiled.ResourcePatterns = append(compiled.ResourcePatterns, pattern)
 
-		matcher, err := NewWildcardMatcher(resource)
+		matcher, err := wildcard.NewWildcardMatcher(resource)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create resource matcher %s: %v", resource, err)
 		}
@@ -377,7 +378,7 @@ func compileStatement(stmt *PolicyStatement) (*CompiledStatement, error) {
 			}
 			compiled.PrincipalPatterns = append(compiled.PrincipalPatterns, pattern)
 
-			matcher, err := NewWildcardMatcher(principal)
+			matcher, err := wildcard.NewWildcardMatcher(principal)
 			if err != nil {
 				return nil, fmt.Errorf("failed to create principal matcher %s: %v", principal, err)
 			}
@@ -403,7 +404,7 @@ func compileStatement(stmt *PolicyStatement) (*CompiledStatement, error) {
 			}
 			compiled.NotResourcePatterns = append(compiled.NotResourcePatterns, pattern)
 
-			matcher, err := NewWildcardMatcher(notResource)
+			matcher, err := wildcard.NewWildcardMatcher(notResource)
 			if err != nil {
 				return nil, fmt.Errorf("failed to create NotResource matcher %s: %v", notResource, err)
 			}
@@ -419,7 +420,7 @@ func compileStatement(stmt *PolicyStatement) (*CompiledStatement, error) {
 
 // compilePattern compiles a wildcard pattern to regex
 func compilePattern(pattern string) (*regexp.Regexp, error) {
-	return CompileWildcardPattern(pattern)
+	return wildcard.CompileWildcardPattern(pattern)
 }
 
 // normalizeToStringSlice converts various types to string slice - kept for backward compatibility
@@ -571,11 +572,11 @@ func (cp *CompiledPolicy) EvaluatePolicy(args *PolicyEvaluationArgs) (bool, Poli
 
 // FastMatchesWildcard uses cached WildcardMatcher for performance
 func FastMatchesWildcard(pattern, str string) bool {
-	matcher, err := GetCachedWildcardMatcher(pattern)
+	matcher, err := wildcard.GetCachedWildcardMatcher(pattern)
 	if err != nil {
 		glog.Errorf("Error getting cached WildcardMatcher for pattern %s: %v", pattern, err)
 		// Fall back to the original implementation
-		return MatchesWildcard(pattern, str)
+		return wildcard.MatchesWildcard(pattern, str)
 	}
 	return matcher.Match(str)
 }

weed/s3api/policy_engine/wildcard_matcher_test.go

@@ -1,469 +0,0 @@
-package policy_engine
-
-import (
-	"testing"
-)
-
-func TestMatchesWildcard(t *testing.T) {
-	tests := []struct {
-		name     string
-		pattern  string
-		str      string
-		expected bool
-	}{
-		// Basic functionality tests
-		{
-			name:     "Exact match",
-			pattern:  "test",
-			str:      "test",
-			expected: true,
-		},
-		{
-			name:     "Single wildcard",
-			pattern:  "*",
-			str:      "anything",
-			expected: true,
-		},
-		{
-			name:     "Empty string with wildcard",
-			pattern:  "*",
-			str:      "",
-			expected: true,
-		},
-
-		// Star (*) wildcard tests
-		{
-			name:     "Prefix wildcard",
-			pattern:  "test*",
-			str:      "test123",
-			expected: true,
-		},
-		{
-			name:     "Suffix wildcard",
-			pattern:  "*test",
-			str:      "123test",
-			expected: true,
-		},
-		{
-			name:     "Middle wildcard",
-			pattern:  "test*123",
-			str:      "testABC123",
-			expected: true,
-		},
-		{
-			name:     "Multiple wildcards",
-			pattern:  "test*abc*123",
-			str:      "testXYZabcDEF123",
-			expected: true,
-		},
-		{
-			name:     "No match",
-			pattern:  "test*",
-			str:      "other",
-			expected: false,
-		},
-
-		// Question mark (?) wildcard tests
-		{
-			name:     "Single question mark",
-			pattern:  "test?",
-			str:      "test1",
-			expected: true,
-		},
-		{
-			name:     "Multiple question marks",
-			pattern:  "test??",
-			str:      "test12",
-			expected: true,
-		},
-		{
-			name:     "Question mark no match",
-			pattern:  "test?",
-			str:      "test12",
-			expected: false,
-		},
-		{
-			name:     "Mixed wildcards",
-			pattern:  "test*abc?def",
-			str:      "testXYZabc1def",
-			expected: true,
-		},
-
-		// Edge cases
-		{
-			name:     "Empty pattern",
-			pattern:  "",
-			str:      "",
-			expected: true,
-		},
-		{
-			name:     "Empty pattern with string",
-			pattern:  "",
-			str:      "test",
-			expected: false,
-		},
-		{
-			name:     "Pattern with string empty",
-			pattern:  "test",
-			str:      "",
-			expected: false,
-		},
-
-		// Special characters
-		{
-			name:     "Pattern with regex special chars",
-			pattern:  "test[abc]",
-			str:      "test[abc]",
-			expected: true,
-		},
-		{
-			name:     "Pattern with dots",
-			pattern:  "test.txt",
-			str:      "test.txt",
-			expected: true,
-		},
-		{
-			name:     "Pattern with dots and wildcard",
-			pattern:  "*.txt",
-			str:      "test.txt",
-			expected: true,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result := MatchesWildcard(tt.pattern, tt.str)
-			if result != tt.expected {
-				t.Errorf("Pattern %s against %s: expected %v, got %v", tt.pattern, tt.str, tt.expected, result)
-			}
-		})
-	}
-}
-
-func TestWildcardMatcher(t *testing.T) {
-	tests := []struct {
-		name     string
-		pattern  string
-		strings  []string
-		expected []bool
-	}{
-		{
-			name:     "Simple star pattern",
-			pattern:  "test*",
-			strings:  []string{"test", "test123", "testing", "other"},
-			expected: []bool{true, true, true, false},
-		},
-		{
-			name:     "Question mark pattern",
-			pattern:  "test?",
-			strings:  []string{"test1", "test2", "test", "test12"},
-			expected: []bool{true, true, false, false},
-		},
-		{
-			name:     "Mixed pattern",
-			pattern:  "*.txt",
-			strings:  []string{"file.txt", "test.txt", "file.doc", "txt"},
-			expected: []bool{true, true, false, false},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			matcher, err := NewWildcardMatcher(tt.pattern)
-			if err != nil {
-				t.Fatalf("Failed to create matcher: %v", err)
-			}
-
-			for i, str := range tt.strings {
-				result := matcher.Match(str)
-				if result != tt.expected[i] {
-					t.Errorf("Pattern %s against %s: expected %v, got %v", tt.pattern, str, tt.expected[i], result)
-				}
-			}
-		})
-	}
-}
-
-func TestCompileWildcardPattern(t *testing.T) {
-	tests := []struct {
-		name    string
-		pattern string
-		input   string
-		want    bool
-	}{
-		{"Star wildcard", "s3:Get*", "s3:GetObject", true},
-		{"Question mark wildcard", "s3:Get?bject", "s3:GetObject", true},
-		{"Mixed wildcards", "s3:*Object*", "s3:GetObjectAcl", true},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			regex, err := CompileWildcardPattern(tt.pattern)
-			if err != nil {
-				t.Errorf("CompileWildcardPattern() error = %v", err)
-				return
-			}
-			got := regex.MatchString(tt.input)
-			if got != tt.want {
-				t.Errorf("CompileWildcardPattern() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-// BenchmarkWildcardMatchingPerformance demonstrates the performance benefits of caching
-func BenchmarkWildcardMatchingPerformance(b *testing.B) {
-	patterns := []string{
-		"s3:Get*",
-		"s3:Put*",
-		"s3:Delete*",
-		"s3:List*",
-		"arn:aws:s3:::bucket/*",
-		"arn:aws:s3:::bucket/prefix*",
-		"user:*",
-		"user:admin-*",
-	}
-
-	inputs := []string{
-		"s3:GetObject",
-		"s3:PutObject",
-		"s3:DeleteObject",
-		"s3:ListBucket",
-		"arn:aws:s3:::bucket/file.txt",
-		"arn:aws:s3:::bucket/prefix/file.txt",
-		"user:admin",
-		"user:admin-john",
-	}
-
-	b.Run("WithoutCache", func(b *testing.B) {
-		for i := 0; i < b.N; i++ {
-			for _, pattern := range patterns {
-				for _, input := range inputs {
-					MatchesWildcard(pattern, input)
-				}
-			}
-		}
-	})
-
-	b.Run("WithCache", func(b *testing.B) {
-		for i := 0; i < b.N; i++ {
-			for _, pattern := range patterns {
-				for _, input := range inputs {
-					FastMatchesWildcard(pattern, input)
-				}
-			}
-		}
-	})
-}
-
-// BenchmarkWildcardMatcherReuse demonstrates the performance benefits of reusing WildcardMatcher instances
-func BenchmarkWildcardMatcherReuse(b *testing.B) {
-	pattern := "s3:Get*"
-	input := "s3:GetObject"
-
-	b.Run("NewMatcherEveryTime", func(b *testing.B) {
-		for i := 0; i < b.N; i++ {
-			matcher, _ := NewWildcardMatcher(pattern)
-			matcher.Match(input)
-		}
-	})
-
-	b.Run("CachedMatcher", func(b *testing.B) {
-		for i := 0; i < b.N; i++ {
-			matcher, _ := GetCachedWildcardMatcher(pattern)
-			matcher.Match(input)
-		}
-	})
-}
-
-// TestWildcardMatcherCaching verifies that caching works correctly
-func TestWildcardMatcherCaching(t *testing.T) {
-	pattern := "s3:Get*"
-
-	// Get the first matcher
-	matcher1, err := GetCachedWildcardMatcher(pattern)
-	if err != nil {
-		t.Fatalf("Failed to get cached matcher: %v", err)
-	}
-
-	// Get the second matcher - should be the same instance
-	matcher2, err := GetCachedWildcardMatcher(pattern)
-	if err != nil {
-		t.Fatalf("Failed to get cached matcher: %v", err)
-	}
-
-	// Check that they're the same instance (same pointer)
-	if matcher1 != matcher2 {
-		t.Errorf("Expected same matcher instance, got different instances")
-	}
-
-	// Test that both matchers work correctly
-	testInput := "s3:GetObject"
-	if !matcher1.Match(testInput) {
-		t.Errorf("First matcher failed to match %s", testInput)
-	}
-	if !matcher2.Match(testInput) {
-		t.Errorf("Second matcher failed to match %s", testInput)
-	}
-}
-
-// TestFastMatchesWildcard verifies that the fast matching function works correctly
-func TestFastMatchesWildcard(t *testing.T) {
-	tests := []struct {
-		pattern string
-		input   string
-		want    bool
-	}{
-		{"s3:Get*", "s3:GetObject", true},
-		{"s3:Put*", "s3:GetObject", false},
-		{"arn:aws:s3:::bucket/*", "arn:aws:s3:::bucket/file.txt", true},
-		{"user:admin-*", "user:admin-john", true},
-		{"user:admin-*", "user:guest-john", false},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.pattern+"_"+tt.input, func(t *testing.T) {
-			got := FastMatchesWildcard(tt.pattern, tt.input)
-			if got != tt.want {
-				t.Errorf("FastMatchesWildcard(%q, %q) = %v, want %v", tt.pattern, tt.input, got, tt.want)
-			}
-		})
-	}
-}
-
-// TestWildcardMatcherCacheBounding tests the bounded cache functionality
-func TestWildcardMatcherCacheBounding(t *testing.T) {
-	// Clear cache before test
-	wildcardMatcherCache.ClearCache()
-
-	// Get original max size
-	originalMaxSize := wildcardMatcherCache.maxSize
-
-	// Set a small max size for testing
-	wildcardMatcherCache.maxSize = 3
-	defer func() {
-		wildcardMatcherCache.maxSize = originalMaxSize
-		wildcardMatcherCache.ClearCache()
-	}()
-
-	// Add patterns up to max size
-	patterns := []string{"pattern1", "pattern2", "pattern3"}
-	for _, pattern := range patterns {
-		_, err := GetCachedWildcardMatcher(pattern)
-		if err != nil {
-			t.Fatalf("Failed to get cached matcher for %s: %v", pattern, err)
-		}
-	}
-
-	// Verify cache size
-	size, maxSize := wildcardMatcherCache.GetCacheStats()
-	if size != 3 {
-		t.Errorf("Expected cache size 3, got %d", size)
-	}
-	if maxSize != 3 {
-		t.Errorf("Expected max size 3, got %d", maxSize)
-	}
-
-	// Add another pattern, should evict the least recently used
-	_, err := GetCachedWildcardMatcher("pattern4")
-	if err != nil {
-		t.Fatalf("Failed to get cached matcher for pattern4: %v", err)
-	}
-
-	// Cache should still be at max size
-	size, _ = wildcardMatcherCache.GetCacheStats()
-	if size != 3 {
-		t.Errorf("Expected cache size 3 after eviction, got %d", size)
-	}
-
-	// The first pattern should have been evicted
-	wildcardMatcherCache.mu.RLock()
-	if _, exists := wildcardMatcherCache.matchers["pattern1"]; exists {
-		t.Errorf("Expected pattern1 to be evicted, but it still exists")
-	}
-	if _, exists := wildcardMatcherCache.matchers["pattern4"]; !exists {
-		t.Errorf("Expected pattern4 to be in cache, but it doesn't exist")
-	}
-	wildcardMatcherCache.mu.RUnlock()
-}
-
-// TestWildcardMatcherCacheLRU tests the LRU eviction policy
-func TestWildcardMatcherCacheLRU(t *testing.T) {
-	// Clear cache before test
-	wildcardMatcherCache.ClearCache()
-
-	// Get original max size
-	originalMaxSize := wildcardMatcherCache.maxSize
-
-	// Set a small max size for testing
-	wildcardMatcherCache.maxSize = 3
-	defer func() {
-		wildcardMatcherCache.maxSize = originalMaxSize
-		wildcardMatcherCache.ClearCache()
-	}()
-
-	// Add patterns to fill cache
-	patterns := []string{"pattern1", "pattern2", "pattern3"}
-	for _, pattern := range patterns {
-		_, err := GetCachedWildcardMatcher(pattern)
-		if err != nil {
-			t.Fatalf("Failed to get cached matcher for %s: %v", pattern, err)
-		}
-	}
-
-	// Access pattern1 to make it most recently used
-	_, err := GetCachedWildcardMatcher("pattern1")
-	if err != nil {
-		t.Fatalf("Failed to access pattern1: %v", err)
-	}
-
-	// Add another pattern, should evict pattern2 (now least recently used)
-	_, err = GetCachedWildcardMatcher("pattern4")
-	if err != nil {
-		t.Fatalf("Failed to get cached matcher for pattern4: %v", err)
-	}
-
-	// pattern1 should still be in cache (was accessed recently)
-	// pattern2 should be evicted (was least recently used)
-	wildcardMatcherCache.mu.RLock()
-	if _, exists := wildcardMatcherCache.matchers["pattern1"]; !exists {
-		t.Errorf("Expected pattern1 to remain in cache (most recently used)")
-	}
-	if _, exists := wildcardMatcherCache.matchers["pattern2"]; exists {
-		t.Errorf("Expected pattern2 to be evicted (least recently used)")
-	}
-	if _, exists := wildcardMatcherCache.matchers["pattern3"]; !exists {
-		t.Errorf("Expected pattern3 to remain in cache")
-	}
-	if _, exists := wildcardMatcherCache.matchers["pattern4"]; !exists {
-		t.Errorf("Expected pattern4 to be in cache")
-	}
-	wildcardMatcherCache.mu.RUnlock()
-}
-
-// TestWildcardMatcherCacheClear tests the cache clearing functionality
-func TestWildcardMatcherCacheClear(t *testing.T) {
-	// Add some patterns to cache
-	patterns := []string{"pattern1", "pattern2", "pattern3"}
-	for _, pattern := range patterns {
-		_, err := GetCachedWildcardMatcher(pattern)
-		if err != nil {
-			t.Fatalf("Failed to get cached matcher for %s: %v", pattern, err)
-		}
-	}
-
-	// Verify cache has patterns
-	size, _ := wildcardMatcherCache.GetCacheStats()
-	if size == 0 {
-		t.Errorf("Expected cache to have patterns before clearing")
-	}
-
-	// Clear cache
-	wildcardMatcherCache.ClearCache()
-
-	// Verify cache is empty
-	size, _ = wildcardMatcherCache.GetCacheStats()
-	if size != 0 {
-		t.Errorf("Expected cache to be empty after clearing, got size %d", size)
-	}
-}

weed/s3api/s3api_server.go

@@ -58,6 +58,7 @@ type S3ApiServerOption struct {
 	Cipher                    bool // encrypt data on volume servers
 	BindIp                    string
 	GrpcPort                  int
+	ExternalUrl               string // external URL clients use, for signature verification behind a reverse proxy
 }
 
 type S3ApiServer struct {

weed/s3api/s3tables/handler_table.go

@@ -1,7 +1,6 @@
 package s3tables
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -10,7 +9,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 )
 
@@ -242,10 +240,6 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 			}
 		}
 
-		if err := h.updateTableLocationMapping(r.Context(), client, "", req.MetadataLocation, tablePath); err != nil {
-			glog.V(1).Infof("failed to update table location mapping for %s: %v", req.MetadataLocation, err)
-		}
-
 		return nil
 	})
 
@@ -499,7 +493,7 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			if err == nil {
 				namespacePolicy = string(policyData)
 			} else if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to fetch namespace policy: %v", err)
+				return fmt.Errorf("failed to fetch namespace policy: %w", err)
 			}
 
 			// Fetch bucket metadata and policy
@@ -509,17 +503,17 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 					return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
 				}
 			} else if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to fetch bucket metadata: %v", err)
+				return fmt.Errorf("failed to fetch bucket metadata: %w", err)
 			}
 
 			policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
 			if err == nil {
 				bucketPolicy = string(policyData)
 			} else if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to fetch bucket policy: %v", err)
+				return fmt.Errorf("failed to fetch bucket policy: %w", err)
 			}
 			if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
-				return err
+				return fmt.Errorf("failed to read bucket tags: %w", err)
 			} else if tags != nil {
 				bucketTags = tags
 			}
@@ -545,6 +539,9 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			}
 
 			tables, paginationToken, err = h.listTablesInNamespaceWithClient(r, client, bucketName, namespaceName, req.Prefix, req.ContinuationToken, maxTables)
+			if err != nil {
+				return fmt.Errorf("list tables in namespace %v: %w", namespaceName, err)
+			}
 		} else {
 			// List tables across all namespaces in bucket
 			bucketPath := GetTableBucketPath(bucketName)
@@ -555,10 +552,10 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			// Fetch bucket metadata and policy
 			data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
 			if err != nil {
-				return err
+				return fmt.Errorf("failed to fetch bucket metadata: %w", err)
 			}
 			if err := json.Unmarshal(data, &bucketMeta); err != nil {
-				return err
+				return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
 			}
 
 			// Fetch bucket policy if it exists
@@ -566,10 +563,10 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			if err == nil {
 				bucketPolicy = string(policyData)
 			} else if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to fetch bucket policy: %v", err)
+				return fmt.Errorf("failed to fetch bucket policy: %w", err)
 			}
 			if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
-				return err
+				return fmt.Errorf("failed to read bucket tags: %w", err)
 			} else if tags != nil {
 				bucketTags = tags
 			}
@@ -586,6 +583,9 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			}
 
 			tables, paginationToken, err = h.listTablesInAllNamespaces(r, client, bucketName, req.Prefix, req.ContinuationToken, maxTables)
+			if err != nil {
+				return fmt.Errorf("list tables in all namespaces: %w", err)
+			}
 		}
 		return err
 	})
@@ -768,8 +768,7 @@ func (h *S3TablesHandler) listTablesInAllNamespaces(r *http.Request, client file
 
 			nsTables, nsToken, err := h.listTablesInNamespaceWithClient(r, client, bucketName, namespace, prefix, tableNameFilter, maxTables-len(tables))
 			if err != nil {
-				glog.Warningf("S3Tables: failed to list tables in namespace %s/%s: %v", bucketName, namespace, err)
-				continue
+				return nil, "", fmt.Errorf("list tables in namespace %s: %w", namespace, err)
 			}
 
 			tables = append(tables, nsTables...)
@@ -938,9 +937,6 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 		if err := h.deleteDirectory(r.Context(), client, tablePath); err != nil {
 			return err
 		}
-		if err := h.deleteTableLocationMapping(r.Context(), client, metadata.MetadataLocation, tablePath); err != nil {
-			glog.V(1).Infof("failed to delete table location mapping for %s: %v", metadata.MetadataLocation, err)
-		}
 		return nil
 	})
 
@@ -1085,9 +1081,6 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		return ErrVersionTokenMismatch
 	}
 
-	// Capture old metadata location before mutation for stale mapping cleanup
-	oldMetadataLocation := metadata.MetadataLocation
-
 	// Update metadata
 	if req.Metadata != nil {
 		if metadata.Metadata == nil {
@@ -1126,9 +1119,6 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil {
 			return err
 		}
-		if err := h.updateTableLocationMapping(r.Context(), client, oldMetadataLocation, metadata.MetadataLocation, tablePath); err != nil {
-			glog.V(1).Infof("failed to update table location mapping for %s -> %s: %v", oldMetadataLocation, metadata.MetadataLocation, err)
-		}
 		return nil
 	})
 
@@ -1144,104 +1134,3 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 	})
 	return nil
 }
-
-func (h *S3TablesHandler) updateTableLocationMapping(ctx context.Context, client filer_pb.SeaweedFilerClient, oldMetadataLocation, newMetadataLocation, tablePath string) error {
-	newTableLocationBucket, ok := parseTableLocationBucket(newMetadataLocation)
-	if !ok {
-		return nil
-	}
-	tableBucketPath, ok := tableBucketPathFromTablePath(tablePath)
-	if !ok {
-		return fmt.Errorf("invalid table path for location mapping: %s", tablePath)
-	}
-
-	if err := h.ensureDirectory(ctx, client, GetTableLocationMappingDir()); err != nil {
-		return err
-	}
-	if err := h.ensureTableLocationMappingBucketDir(ctx, client, newTableLocationBucket); err != nil {
-		return err
-	}
-
-	// If the metadata location changed, remove this table's stale mapping entry from the old bucket.
-	if oldMetadataLocation != "" && oldMetadataLocation != newMetadataLocation {
-		oldTableLocationBucket, ok := parseTableLocationBucket(oldMetadataLocation)
-		if ok && oldTableLocationBucket != newTableLocationBucket {
-			if err := h.removeTableLocationMappingEntry(ctx, client, oldTableLocationBucket, tablePath); err != nil {
-				glog.V(1).Infof("failed to delete stale mapping for %s: %v", oldTableLocationBucket, err)
-			}
-		}
-	}
-
-	return h.upsertFile(ctx, client, GetTableLocationMappingEntryPath(newTableLocationBucket, tablePath), []byte(tableBucketPath))
-}
-
-func (h *S3TablesHandler) deleteTableLocationMapping(ctx context.Context, client filer_pb.SeaweedFilerClient, metadataLocation, tablePath string) error {
-	tableLocationBucket, ok := parseTableLocationBucket(metadataLocation)
-	if !ok {
-		return nil
-	}
-	return h.removeTableLocationMappingEntry(ctx, client, tableLocationBucket, tablePath)
-}
-
-func (h *S3TablesHandler) ensureTableLocationMappingBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, tableLocationBucket string) error {
-	mappingDir := GetTableLocationMappingDir()
-	bucketMappingPath := GetTableLocationMappingPath(tableLocationBucket)
-
-	resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
-		Directory: mappingDir,
-		Name:      tableLocationBucket,
-	})
-	if err == nil {
-		if resp != nil && resp.Entry != nil && resp.Entry.IsDirectory {
-			return nil
-		}
-		if removeErr := h.deleteEntryIfExists(ctx, client, bucketMappingPath); removeErr != nil && !errors.Is(removeErr, filer_pb.ErrNotFound) {
-			return removeErr
-		}
-	} else if !errors.Is(err, filer_pb.ErrNotFound) {
-		return err
-	}
-
-	return h.ensureDirectory(ctx, client, bucketMappingPath)
-}
-
-func (h *S3TablesHandler) removeTableLocationMappingEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, tableLocationBucket, tablePath string) error {
-	entryPath := GetTableLocationMappingEntryPath(tableLocationBucket, tablePath)
-	if err := h.deleteEntryIfExists(ctx, client, entryPath); err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
-		return err
-	}
-	return h.removeTableLocationMappingBucketDirIfEmpty(ctx, client, tableLocationBucket)
-}
-
-func (h *S3TablesHandler) removeTableLocationMappingBucketDirIfEmpty(ctx context.Context, client filer_pb.SeaweedFilerClient, tableLocationBucket string) error {
-	bucketMappingPath := GetTableLocationMappingPath(tableLocationBucket)
-
-	stream, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
-		Directory: bucketMappingPath,
-		Limit:     1,
-	})
-	if err != nil {
-		if errors.Is(err, filer_pb.ErrNotFound) {
-			return nil
-		}
-		return err
-	}
-
-	for {
-		resp, recvErr := stream.Recv()
-		if recvErr == io.EOF {
-			break
-		}
-		if recvErr != nil {
-			return recvErr
-		}
-		if resp != nil && resp.Entry != nil {
-			return nil
-		}
-	}
-
-	if err := h.deleteEntryIfExists(ctx, client, bucketMappingPath); err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
-		return err
-	}
-	return nil
-}

weed/s3api/s3tables/permissions.go

@@ -8,6 +8,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // Permission represents a specific action permission
@@ -217,7 +218,7 @@ func hasIdentityPermission(operation string, ctx *PolicyContext) bool {
 			if action == candidate {
 				return true
 			}
-			if strings.ContainsAny(action, "*?") && policy_engine.MatchesWildcard(action, candidate) {
+			if strings.ContainsAny(action, "*?") && wildcard.MatchesWildcard(action, candidate) {
 				return true
 			}
 		}
@@ -238,7 +239,7 @@ func matchesPrincipal(principalSpec interface{}, principal string) bool {
 			return true
 		}
 		// Support wildcard matching for principals (e.g., "arn:aws:iam::*:user/admin")
-		return policy_engine.MatchesWildcard(p, principal)
+		return wildcard.MatchesWildcard(p, principal)
 	case []interface{}:
 		// Array of principals
 		for _, item := range p {
@@ -247,7 +248,7 @@ func matchesPrincipal(principalSpec interface{}, principal string) bool {
 					return true
 				}
 				// Support wildcard matching
-				if policy_engine.MatchesWildcard(str, principal) {
+				if wildcard.MatchesWildcard(str, principal) {
 					return true
 				}
 			}
@@ -302,7 +303,7 @@ func matchesActionPattern(pattern, action string) bool {
 
 	// Wildcard match using policy engine's wildcard matcher
 	// Supports both * (any sequence) and ? (single character) anywhere in the pattern
-	return policy_engine.MatchesWildcard(pattern, action)
+	return wildcard.MatchesWildcard(pattern, action)
 }
 
 func matchesConditions(conditions map[string]map[string]interface{}, ctx *PolicyContext) bool {
@@ -411,7 +412,7 @@ func matchesResourcePattern(pattern, resourceARN string) bool {
 	}
 
 	// Wildcard match using policy engine's wildcard matcher
-	return policy_engine.MatchesWildcard(pattern, resourceARN)
+	return wildcard.MatchesWildcard(pattern, resourceARN)
 }
 
 // Helper functions for specific permissions

weed/s3api/s3tables/table_location_mapping_test.go

@@ -1,77 +0,0 @@
-package s3tables
-
-import (
-	"strings"
-	"testing"
-)
-
-func TestGetTableLocationMappingEntryPathPerTable(t *testing.T) {
-	tableLocationBucket := "shared-location--table-s3"
-	tablePathA := GetTablePath("warehouse", "analytics", "orders")
-	tablePathB := GetTablePath("warehouse", "analytics", "customers")
-
-	entryPathA := GetTableLocationMappingEntryPath(tableLocationBucket, tablePathA)
-	entryPathARepeat := GetTableLocationMappingEntryPath(tableLocationBucket, tablePathA)
-	entryPathB := GetTableLocationMappingEntryPath(tableLocationBucket, tablePathB)
-
-	if entryPathA != entryPathARepeat {
-		t.Fatalf("mapping entry path should be deterministic: %q != %q", entryPathA, entryPathARepeat)
-	}
-	if entryPathA == entryPathB {
-		t.Fatalf("mapping entry path should differ per table path: %q == %q", entryPathA, entryPathB)
-	}
-
-	expectedPrefix := GetTableLocationMappingPath(tableLocationBucket) + "/"
-	if !strings.HasPrefix(entryPathA, expectedPrefix) {
-		t.Fatalf("mapping entry path %q should start with %q", entryPathA, expectedPrefix)
-	}
-	if strings.TrimPrefix(entryPathA, expectedPrefix) == "" {
-		t.Fatalf("mapping entry name should not be empty: %q", entryPathA)
-	}
-}
-
-func TestTableBucketPathFromTablePath(t *testing.T) {
-	testCases := []struct {
-		name      string
-		tablePath string
-		expected  string
-		ok        bool
-	}{
-		{
-			name:      "valid table path",
-			tablePath: GetTablePath("warehouse", "analytics", "orders"),
-			expected:  GetTableBucketPath("warehouse"),
-			ok:        true,
-		},
-		{
-			name:      "valid table bucket root",
-			tablePath: GetTableBucketPath("warehouse"),
-			expected:  GetTableBucketPath("warehouse"),
-			ok:        true,
-		},
-		{
-			name:      "invalid non-tables path",
-			tablePath: "/tmp/warehouse/analytics/orders",
-			expected:  "",
-			ok:        false,
-		},
-		{
-			name:      "invalid empty bucket segment",
-			tablePath: "/buckets/",
-			expected:  "",
-			ok:        false,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			actual, ok := tableBucketPathFromTablePath(tc.tablePath)
-			if ok != tc.ok {
-				t.Fatalf("tableBucketPathFromTablePath(%q) ok=%v, want %v", tc.tablePath, ok, tc.ok)
-			}
-			if actual != tc.expected {
-				t.Fatalf("tableBucketPathFromTablePath(%q)=%q, want %q", tc.tablePath, actual, tc.expected)
-			}
-		})
-	}
-}

weed/s3api/s3tables/utils.go

@@ -2,7 +2,6 @@ package s3tables
 
 import (
 	"crypto/rand"
-	"crypto/sha1"
 	"encoding/hex"
 	"fmt"
 	"net/url"
@@ -21,8 +20,7 @@ const (
 )
 
 const (
-	tableLocationMappingsDirName = ".table-location-mappings"
-	tableObjectRootDirName       = ".objects"
+	tableObjectRootDirName = ".objects"
 )
 
 var (
@@ -109,43 +107,6 @@ func GetTableObjectBucketPath(bucketName string) string {
 	return path.Join(GetTableObjectRootDir(), bucketName)
 }
 
-// GetTableLocationMappingDir returns the root path for table location bucket mappings
-func GetTableLocationMappingDir() string {
-	return path.Join(TablesPath, tableLocationMappingsDirName)
-}
-
-// GetTableLocationMappingPath returns the filer path for a table location bucket mapping
-func GetTableLocationMappingPath(tableLocationBucket string) string {
-	return path.Join(GetTableLocationMappingDir(), tableLocationBucket)
-}
-
-// GetTableLocationMappingEntryPath returns the filer path for a table-specific mapping entry.
-// Each table gets its own entry so multiple tables can share the same external table-location bucket.
-func GetTableLocationMappingEntryPath(tableLocationBucket, tablePath string) string {
-	return path.Join(GetTableLocationMappingPath(tableLocationBucket), tableLocationMappingEntryName(tablePath))
-}
-
-func tableLocationMappingEntryName(tablePath string) string {
-	normalized := path.Clean("/" + strings.TrimSpace(strings.TrimPrefix(tablePath, "/")))
-	sum := sha1.Sum([]byte(normalized))
-	return hex.EncodeToString(sum[:])
-}
-
-func tableBucketPathFromTablePath(tablePath string) (string, bool) {
-	normalized := path.Clean("/" + strings.TrimSpace(strings.TrimPrefix(tablePath, "/")))
-	tablesPrefix := strings.TrimSuffix(TablesPath, "/") + "/"
-	if !strings.HasPrefix(normalized, tablesPrefix) {
-		return "", false
-	}
-
-	remaining := strings.TrimPrefix(normalized, tablesPrefix)
-	bucketName, _, _ := strings.Cut(remaining, "/")
-	if bucketName == "" {
-		return "", false
-	}
-	return path.Join(TablesPath, bucketName), true
-}
-
 // Metadata structures
 
 type tableBucketMetadata struct {
@@ -244,22 +205,6 @@ func ValidateBucketName(name string) error {
 	return validateBucketName(name)
 }
 
-func parseTableLocationBucket(metadataLocation string) (string, bool) {
-	if !strings.HasPrefix(metadataLocation, "s3://") {
-		return "", false
-	}
-	trimmed := strings.TrimPrefix(metadataLocation, "s3://")
-	trimmed = strings.TrimSuffix(trimmed, "/")
-	if trimmed == "" {
-		return "", false
-	}
-	bucket, _, _ := strings.Cut(trimmed, "/")
-	if bucket == "" || !strings.HasSuffix(bucket, "--table-s3") {
-		return "", false
-	}
-	return bucket, true
-}
-
 // BuildBucketARN builds a bucket ARN with the provided region and account ID.
 // If region is empty, the ARN will omit the region field.
 func BuildBucketARN(region, accountID, bucketName string) (string, error) {

weed/shell/command_s3_bucket_versioning.go

@@ -0,0 +1,167 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3BucketVersioning{})
+}
+
+type commandS3BucketVersioning struct {
+}
+
+func (c *commandS3BucketVersioning) Name() string {
+	return "s3.bucket.versioning"
+}
+
+func (c *commandS3BucketVersioning) Help() string {
+	return `view or update S3 bucket versioning configuration
+
+	Example:
+		# View the current versioning status
+		s3.bucket.versioning -name <bucket_name>
+
+		# Enable versioning
+		s3.bucket.versioning -name <bucket_name> -enable
+
+		# Suspend versioning
+		s3.bucket.versioning -name <bucket_name> -suspend
+
+		# Set versioning status explicitly (Enabled or Suspended)
+		s3.bucket.versioning -name <bucket_name> -status Enabled
+
+	Object Lock requires versioning to be enabled and prevents suspending it.
+`
+}
+
+func (c *commandS3BucketVersioning) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3BucketVersioning) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+	bucketCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	bucketName := bucketCommand.String("name", "", "bucket name")
+	statusFlag := bucketCommand.String("status", "", "versioning status: Enabled or Suspended")
+	enable := bucketCommand.Bool("enable", false, "enable versioning on the bucket")
+	suspend := bucketCommand.Bool("suspend", false, "suspend versioning on the bucket")
+	if err = bucketCommand.Parse(args); err != nil {
+		return err
+	}
+
+	if *bucketName == "" {
+		return fmt.Errorf("empty bucket name")
+	}
+
+	desiredStatus, err := normalizeVersioningStatus(*statusFlag, *enable, *suspend)
+	if err != nil {
+		return err
+	}
+
+	err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		if err != nil {
+			return fmt.Errorf("get filer configuration: %w", err)
+		}
+		filerBucketsPath := resp.DirBuckets
+
+		lookupResp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+			Directory: filerBucketsPath,
+			Name:      *bucketName,
+		})
+		if err != nil {
+			return fmt.Errorf("lookup bucket %s: %w", *bucketName, err)
+		}
+
+		entry := lookupResp.Entry
+		currentStatus, lockEnabled := getBucketVersioningState(entry)
+
+		if desiredStatus == "" {
+			if lockEnabled {
+				currentStatus = s3_constants.VersioningEnabled
+			}
+			fmt.Fprintf(writer, "Bucket: %s\n", *bucketName)
+			switch currentStatus {
+			case s3_constants.VersioningEnabled:
+				fmt.Fprintf(writer, "Versioning: Enabled\n")
+			case s3_constants.VersioningSuspended:
+				fmt.Fprintf(writer, "Versioning: Suspended\n")
+			case "":
+				fmt.Fprintf(writer, "Versioning: Unconfigured\n")
+			default:
+				fmt.Fprintf(writer, "Versioning: %s\n", currentStatus)
+			}
+			if lockEnabled {
+				fmt.Fprintf(writer, "Object Lock: Enabled\n")
+			}
+			return nil
+		}
+
+		if lockEnabled && desiredStatus == s3_constants.VersioningSuspended {
+			return fmt.Errorf("cannot suspend versioning on bucket %s: Object Lock is enabled", *bucketName)
+		}
+
+		if entry.Extended == nil {
+			entry.Extended = make(map[string][]byte)
+		}
+		entry.Extended[s3_constants.ExtVersioningKey] = []byte(desiredStatus)
+
+		if _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
+			Directory: filerBucketsPath,
+			Entry:     entry,
+		}); err != nil {
+			return fmt.Errorf("failed to update bucket: %w", err)
+		}
+
+		fmt.Fprintf(writer, "Bucket %s versioning set to %s\n", *bucketName, desiredStatus)
+		return nil
+	})
+
+	return err
+}
+
+func normalizeVersioningStatus(statusFlag string, enable bool, suspend bool) (string, error) {
+	if enable && suspend {
+		return "", fmt.Errorf("only one of -enable or -suspend can be set")
+	}
+	if (enable || suspend) && statusFlag != "" {
+		return "", fmt.Errorf("use either -status or -enable/-suspend, not both")
+	}
+	if enable {
+		return s3_constants.VersioningEnabled, nil
+	}
+	if suspend {
+		return s3_constants.VersioningSuspended, nil
+	}
+	if statusFlag == "" {
+		return "", nil
+	}
+	switch strings.ToLower(strings.TrimSpace(statusFlag)) {
+	case "enabled":
+		return s3_constants.VersioningEnabled, nil
+	case "suspended":
+		return s3_constants.VersioningSuspended, nil
+	default:
+		return "", fmt.Errorf("invalid versioning status %q: must be Enabled or Suspended", statusFlag)
+	}
+}
+
+func getBucketVersioningState(entry *filer_pb.Entry) (status string, lockEnabled bool) {
+	if entry.Extended == nil {
+		return "", false
+	}
+	if versioning, ok := entry.Extended[s3_constants.ExtVersioningKey]; ok {
+		status = string(versioning)
+	}
+	if lockStatus, ok := entry.Extended[s3_constants.ExtObjectLockEnabledKey]; ok {
+		lockEnabled = string(lockStatus) == s3_constants.ObjectLockEnabled
+	}
+	return status, lockEnabled
+}

weed/storage/erasure_coding/shard_distribution.go

@@ -0,0 +1,367 @@
+package erasure_coding
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"reflect"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	"google.golang.org/grpc"
+)
+
+func ensureLogger(logger logger) logger {
+	if logger == nil {
+		return &glogFallbackLogger{}
+	}
+	return logger
+}
+
+type logger interface {
+	Info(string, ...interface{})
+	Warning(string, ...interface{})
+	Error(string, ...interface{})
+}
+
+type withFieldLogger interface {
+	WithFields(map[string]interface{}) logger
+}
+
+func withFields(log logger, fields map[string]interface{}) logger {
+	if fields == nil {
+		return log
+	}
+	if wf, ok := log.(withFieldLogger); ok {
+		return wf.WithFields(fields)
+	}
+
+	val := reflect.ValueOf(log)
+	method := val.MethodByName("WithFields")
+	if !method.IsValid() {
+		return log
+	}
+	methodType := method.Type()
+	if methodType.NumIn() != 1 || methodType.NumOut() != 1 {
+		return log
+	}
+	argType := methodType.In(0)
+	mapType := reflect.TypeOf(map[string]interface{}{})
+	var arg reflect.Value
+	if mapType.AssignableTo(argType) {
+		arg = reflect.ValueOf(fields)
+	} else if mapType.ConvertibleTo(argType) {
+		arg = reflect.ValueOf(fields).Convert(argType)
+	} else {
+		return log
+	}
+	result := method.Call([]reflect.Value{arg})[0]
+	if result.IsValid() && result.CanInterface() {
+		if enhanced, ok := result.Interface().(logger); ok {
+			return enhanced
+		}
+	}
+	return log
+}
+
+type glogFallbackLogger struct{}
+
+func (g *glogFallbackLogger) Info(msg string, args ...interface{}) {
+	if len(args) > 0 {
+		glog.Infof(msg, args...)
+		return
+	}
+	glog.Info(msg)
+}
+
+func (g *glogFallbackLogger) Warning(msg string, args ...interface{}) {
+	if len(args) > 0 {
+		glog.Warningf(msg, args...)
+		return
+	}
+	glog.Warning(msg)
+}
+
+func (g *glogFallbackLogger) Error(msg string, args ...interface{}) {
+	if len(args) > 0 {
+		glog.Errorf(msg, args...)
+		return
+	}
+	glog.Error(msg)
+}
+
+// DistributeEcShards distributes locally generated EC shards to destination servers.
+// Returns the shard assignment map used for mounting.
+func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb.TaskTarget, shardFiles map[string]string, dialOption grpc.DialOption, logger logger) (map[string][]string, error) {
+	if len(targets) == 0 {
+		return nil, fmt.Errorf("no targets specified for EC shard distribution")
+	}
+
+	if len(shardFiles) == 0 {
+		return nil, fmt.Errorf("no shard files available for distribution")
+	}
+
+	log := ensureLogger(logger)
+
+	shardAssignment := make(map[string][]string)
+
+	for _, target := range targets {
+		if len(target.ShardIds) == 0 {
+			continue
+		}
+
+		var assignedShards []string
+		for _, shardId := range target.ShardIds {
+			shardType := fmt.Sprintf("ec%02d", shardId)
+			assignedShards = append(assignedShards, shardType)
+		}
+
+		if len(assignedShards) > 0 {
+			if _, hasEcx := shardFiles["ecx"]; hasEcx {
+				assignedShards = append(assignedShards, "ecx")
+			}
+			if _, hasEcj := shardFiles["ecj"]; hasEcj {
+				assignedShards = append(assignedShards, "ecj")
+			}
+			if _, hasVif := shardFiles["vif"]; hasVif {
+				assignedShards = append(assignedShards, "vif")
+			}
+		}
+
+		existing := shardAssignment[target.Node]
+		if len(existing) == 0 {
+			shardAssignment[target.Node] = assignedShards
+			continue
+		}
+
+		seen := make(map[string]struct{}, len(existing))
+		for _, shard := range existing {
+			seen[shard] = struct{}{}
+		}
+		for _, shard := range assignedShards {
+			if _, ok := seen[shard]; ok {
+				continue
+			}
+			seen[shard] = struct{}{}
+			existing = append(existing, shard)
+		}
+		shardAssignment[target.Node] = existing
+	}
+
+	if len(shardAssignment) == 0 {
+		return nil, fmt.Errorf("no shard assignments found from planning phase")
+	}
+
+	for destNode, assignedShards := range shardAssignment {
+		withFields(log, map[string]interface{}{
+			"destination":     destNode,
+			"assigned_shards": len(assignedShards),
+			"shard_types":     assignedShards,
+		}).Info("Starting shard distribution to destination server")
+
+		var transferredBytes int64
+		for _, shardType := range assignedShards {
+			filePath, exists := shardFiles[shardType]
+			if !exists {
+				return nil, fmt.Errorf("shard file %s not found for destination %s", shardType, destNode)
+			}
+
+			if info, err := os.Stat(filePath); err == nil {
+				transferredBytes += info.Size()
+				withFields(log, map[string]interface{}{
+					"destination": destNode,
+					"shard_type":  shardType,
+					"file_path":   filePath,
+					"size_bytes":  info.Size(),
+					"size_kb":     float64(info.Size()) / 1024,
+				}).Info("Starting shard file transfer")
+			}
+
+			if err := sendShardFileToDestination(volumeID, collection, dialOption, destNode, filePath, shardType); err != nil {
+				return nil, fmt.Errorf("failed to send %s to %s: %w", shardType, destNode, err)
+			}
+
+			withFields(log, map[string]interface{}{
+				"destination": destNode,
+				"shard_type":  shardType,
+			}).Info("Shard file transfer completed")
+		}
+
+		withFields(log, map[string]interface{}{
+			"destination":        destNode,
+			"shards_transferred": len(assignedShards),
+			"total_bytes":        transferredBytes,
+			"total_mb":           float64(transferredBytes) / (1024 * 1024),
+		}).Info("All shards distributed to destination server")
+	}
+
+	glog.V(1).Infof("Successfully distributed EC shards to %d destinations", len(shardAssignment))
+	return shardAssignment, nil
+}
+
+// MountEcShards mounts EC shards on destination servers using an assignment map.
+func MountEcShards(volumeID uint32, collection string, shardAssignment map[string][]string, dialOption grpc.DialOption, logger logger) error {
+	if shardAssignment == nil {
+		return fmt.Errorf("shard assignment not available for mounting")
+	}
+
+	log := ensureLogger(logger)
+
+	var mountErrors []error
+	for destNode, assignedShards := range shardAssignment {
+		var shardIds []uint32
+		var metadataFiles []string
+
+		for _, shardType := range assignedShards {
+			if strings.HasPrefix(shardType, "ec") && len(shardType) == 4 {
+				var shardId uint32
+				if _, err := fmt.Sscanf(shardType[2:], "%d", &shardId); err == nil {
+					shardIds = append(shardIds, shardId)
+				}
+			} else {
+				metadataFiles = append(metadataFiles, shardType)
+			}
+		}
+
+		withFields(log, map[string]interface{}{
+			"destination":    destNode,
+			"shard_ids":      shardIds,
+			"shard_count":    len(shardIds),
+			"metadata_files": metadataFiles,
+		}).Info("Starting EC shard mount operation")
+
+		if len(shardIds) == 0 {
+			withFields(log, map[string]interface{}{
+				"destination":    destNode,
+				"metadata_files": metadataFiles,
+			}).Info("No EC shards to mount (only metadata files)")
+			continue
+		}
+
+		err := operation.WithVolumeServerClient(false, pb.ServerAddress(destNode), dialOption,
+			func(client volume_server_pb.VolumeServerClient) error {
+				_, mountErr := client.VolumeEcShardsMount(context.Background(), &volume_server_pb.VolumeEcShardsMountRequest{
+					VolumeId:   volumeID,
+					Collection: collection,
+					ShardIds:   shardIds,
+				})
+				return mountErr
+			})
+
+		if err != nil {
+			mountErrors = append(mountErrors, fmt.Errorf("mount %s shards %v: %w", destNode, shardIds, err))
+			withFields(log, map[string]interface{}{
+				"destination": destNode,
+				"shard_ids":   shardIds,
+				"error":       err.Error(),
+			}).Error("Failed to mount EC shards")
+		} else {
+			withFields(log, map[string]interface{}{
+				"destination": destNode,
+				"shard_ids":   shardIds,
+				"volume_id":   volumeID,
+				"collection":  collection,
+			}).Info("Successfully mounted EC shards")
+		}
+	}
+
+	if len(mountErrors) > 0 {
+		return errors.Join(mountErrors...)
+	}
+	return nil
+}
+
+// sendShardFileToDestination sends a single shard file to a destination server using ReceiveFile API.
+func sendShardFileToDestination(volumeID uint32, collection string, dialOption grpc.DialOption, destServer, filePath, shardType string) error {
+	return operation.WithVolumeServerClient(false, pb.ServerAddress(destServer), dialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			file, err := os.Open(filePath)
+			if err != nil {
+				return fmt.Errorf("failed to open shard file %s: %v", filePath, err)
+			}
+			defer file.Close()
+
+			fileInfo, err := file.Stat()
+			if err != nil {
+				return fmt.Errorf("failed to get file info for %s: %v", filePath, err)
+			}
+
+			var ext string
+			var shardId uint32
+			if shardType == "ecx" {
+				ext = ".ecx"
+				shardId = 0
+			} else if shardType == "ecj" {
+				ext = ".ecj"
+				shardId = 0
+			} else if shardType == "vif" {
+				ext = ".vif"
+				shardId = 0
+			} else if strings.HasPrefix(shardType, "ec") && len(shardType) == 4 {
+				ext = "." + shardType
+				fmt.Sscanf(shardType[2:], "%d", &shardId)
+			} else {
+				return fmt.Errorf("unknown shard type: %s", shardType)
+			}
+
+			stream, err := client.ReceiveFile(context.Background())
+			if err != nil {
+				return fmt.Errorf("failed to create receive stream: %v", err)
+			}
+
+			err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+				Data: &volume_server_pb.ReceiveFileRequest_Info{
+					Info: &volume_server_pb.ReceiveFileInfo{
+						VolumeId:   volumeID,
+						Ext:        ext,
+						Collection: collection,
+						IsEcVolume: true,
+						ShardId:    shardId,
+						FileSize:   uint64(fileInfo.Size()),
+					},
+				},
+			})
+			if err != nil {
+				return fmt.Errorf("failed to send file info: %v", err)
+			}
+
+			buffer := make([]byte, 64*1024)
+			for {
+				n, readErr := file.Read(buffer)
+				if n > 0 {
+					err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+						Data: &volume_server_pb.ReceiveFileRequest_FileContent{
+							FileContent: buffer[:n],
+						},
+					})
+					if err != nil {
+						return fmt.Errorf("failed to send file content: %v", err)
+					}
+				}
+				if readErr == io.EOF {
+					break
+				}
+				if readErr != nil {
+					return fmt.Errorf("failed to read file: %v", readErr)
+				}
+			}
+
+			resp, err := stream.CloseAndRecv()
+			if err != nil {
+				return fmt.Errorf("failed to close stream: %v", err)
+			}
+
+			if resp.Error != "" {
+				return fmt.Errorf("server error: %s", resp.Error)
+			}
+
+			glog.V(2).Infof("Successfully sent %s (%d bytes) to %s", shardType, resp.BytesWritten, destServer)
+			return nil
+		})
+}

weed/util/wildcard/filter.go

@@ -0,0 +1,39 @@
+package wildcard
+
+import "strings"
+
+// CompileWildcardMatchers parses comma-separated wildcard patterns and compiles them.
+// Empty tokens are ignored. Invalid patterns are skipped.
+func CompileWildcardMatchers(filter string) []*WildcardMatcher {
+	parts := strings.Split(filter, ",")
+	matchers := make([]*WildcardMatcher, 0, len(parts))
+	for _, part := range parts {
+		trimmed := strings.TrimSpace(part)
+		if trimmed == "" {
+			continue
+		}
+		matcher, err := NewWildcardMatcher(trimmed)
+		if err != nil {
+			continue
+		}
+		matchers = append(matchers, matcher)
+	}
+	if len(matchers) == 0 {
+		return nil
+	}
+	return matchers
+}
+
+// MatchesAnyWildcard returns true when no matcher is provided,
+// or when any matcher matches the given value.
+func MatchesAnyWildcard(matchers []*WildcardMatcher, value string) bool {
+	if len(matchers) == 0 {
+		return true
+	}
+	for _, matcher := range matchers {
+		if matcher != nil && matcher.Match(value) {
+			return true
+		}
+	}
+	return false
+}

weed/util/wildcard/wildcard_matcher.go

@@ -1,4 +1,4 @@
-package policy_engine
+package wildcard
 
 import (
 	"regexp"
@@ -8,27 +8,25 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 )
 
-// WildcardMatcher provides unified wildcard matching functionality
+// WildcardMatcher provides unified wildcard matching functionality.
 type WildcardMatcher struct {
-	// Use regex for complex patterns with ? wildcards
-	// Use string manipulation for simple * patterns (better performance)
 	useRegex bool
 	regex    *regexp.Regexp
 	pattern  string
 }
 
-// WildcardMatcherCache provides caching for WildcardMatcher instances
+// WildcardMatcherCache provides caching for WildcardMatcher instances.
 type WildcardMatcherCache struct {
 	mu          sync.RWMutex
 	matchers    map[string]*WildcardMatcher
 	maxSize     int
-	accessOrder []string // For LRU eviction
+	accessOrder []string
 }
 
-// NewWildcardMatcherCache creates a new WildcardMatcherCache with a configurable maxSize
+// NewWildcardMatcherCache creates a new WildcardMatcherCache with a configurable maxSize.
 func NewWildcardMatcherCache(maxSize int) *WildcardMatcherCache {
 	if maxSize <= 0 {
-		maxSize = 1000 // Default value
+		maxSize = 1000
 	}
 	return &WildcardMatcherCache{
 		matchers: make(map[string]*WildcardMatcher),
@@ -36,12 +34,10 @@ func NewWildcardMatcherCache(maxSize int) *WildcardMatcherCache {
 	}
 }
 
-// Global cache instance
-var wildcardMatcherCache = NewWildcardMatcherCache(1000) // Default maxSize
+var wildcardMatcherCache = NewWildcardMatcherCache(1000)
 
-// GetCachedWildcardMatcher gets or creates a cached WildcardMatcher for the given pattern
+// GetCachedWildcardMatcher gets or creates a cached WildcardMatcher for the given pattern.
 func GetCachedWildcardMatcher(pattern string) (*WildcardMatcher, error) {
-	// Fast path: check if already in cache
 	wildcardMatcherCache.mu.RLock()
 	if matcher, exists := wildcardMatcherCache.matchers[pattern]; exists {
 		wildcardMatcherCache.mu.RUnlock()
@@ -50,66 +46,53 @@ func GetCachedWildcardMatcher(pattern string) (*WildcardMatcher, error) {
 	}
 	wildcardMatcherCache.mu.RUnlock()
 
-	// Slow path: create new matcher and cache it
 	wildcardMatcherCache.mu.Lock()
 	defer wildcardMatcherCache.mu.Unlock()
 
-	// Double-check after acquiring write lock
 	if matcher, exists := wildcardMatcherCache.matchers[pattern]; exists {
 		wildcardMatcherCache.updateAccessOrderLocked(pattern)
 		return matcher, nil
 	}
 
-	// Create new matcher
 	matcher, err := NewWildcardMatcher(pattern)
 	if err != nil {
 		return nil, err
 	}
 
-	// Evict old entries if cache is full
 	if len(wildcardMatcherCache.matchers) >= wildcardMatcherCache.maxSize {
 		wildcardMatcherCache.evictLeastRecentlyUsed()
 	}
 
-	// Cache it
 	wildcardMatcherCache.matchers[pattern] = matcher
 	wildcardMatcherCache.accessOrder = append(wildcardMatcherCache.accessOrder, pattern)
 	return matcher, nil
 }
 
-// updateAccessOrder updates the access order for LRU eviction (with read lock)
 func (c *WildcardMatcherCache) updateAccessOrder(pattern string) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.updateAccessOrderLocked(pattern)
 }
 
-// updateAccessOrderLocked updates the access order for LRU eviction (without locking)
 func (c *WildcardMatcherCache) updateAccessOrderLocked(pattern string) {
-	// Remove pattern from its current position
 	for i, p := range c.accessOrder {
 		if p == pattern {
 			c.accessOrder = append(c.accessOrder[:i], c.accessOrder[i+1:]...)
 			break
 		}
 	}
-	// Add pattern to the end (most recently used)
 	c.accessOrder = append(c.accessOrder, pattern)
 }
 
-// evictLeastRecentlyUsed removes the least recently used pattern from the cache
 func (c *WildcardMatcherCache) evictLeastRecentlyUsed() {
 	if len(c.accessOrder) == 0 {
 		return
 	}
-
-	// Remove the least recently used pattern (first in the list)
 	lruPattern := c.accessOrder[0]
 	c.accessOrder = c.accessOrder[1:]
 	delete(c.matchers, lruPattern)
 }
 
-// ClearCache clears all cached patterns (useful for testing)
 func (c *WildcardMatcherCache) ClearCache() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -117,26 +100,19 @@ func (c *WildcardMatcherCache) ClearCache() {
 	c.accessOrder = c.accessOrder[:0]
 }
 
-// GetCacheStats returns cache statistics
 func (c *WildcardMatcherCache) GetCacheStats() (size int, maxSize int) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return len(c.matchers), c.maxSize
 }
 
-// NewWildcardMatcher creates a new wildcard matcher for the given pattern
-// The matcher uses an efficient string-based algorithm that handles both * and ? wildcards
-// without requiring regex compilation.
+// NewWildcardMatcher creates a new wildcard matcher for the given pattern.
 func NewWildcardMatcher(pattern string) (*WildcardMatcher, error) {
-	matcher := &WildcardMatcher{
-		pattern:  pattern,
-		useRegex: false, // String-based matching now handles both * and ?
-	}
-
+	matcher := &WildcardMatcher{pattern: pattern, useRegex: false}
 	return matcher, nil
 }
 
-// Match checks if a string matches the wildcard pattern
+// Match checks if a string matches the wildcard pattern.
 func (m *WildcardMatcher) Match(str string) bool {
 	if m.useRegex {
 		return m.regex.MatchString(str)
@@ -144,32 +120,27 @@ func (m *WildcardMatcher) Match(str string) bool {
 	return matchWildcardString(m.pattern, str)
 }
 
-// MatchesWildcard provides a simple function interface for wildcard matching
-// This function consolidates the logic from the previous separate implementations
-//
-// Rules:
-//   - '*' matches any sequence of characters (including empty string)
-//   - '?' matches exactly one character (any character)
+// MatchesWildcard provides a simple function interface for wildcard matching.
 func MatchesWildcard(pattern, str string) bool {
-	// matchWildcardString now handles both * and ? efficiently without regex
 	return matchWildcardString(pattern, str)
 }
 
-// CompileWildcardPattern converts a wildcard pattern to a compiled regex
-// This replaces the previous compilePattern function
+// FastMatchesWildcard uses a cached WildcardMatcher for repeated pattern matching.
+func FastMatchesWildcard(pattern, str string) bool {
+	matcher, err := GetCachedWildcardMatcher(pattern)
+	if err != nil {
+		glog.Errorf("Error getting cached WildcardMatcher for pattern %s: %v. Falling back to MatchesWildcard.", pattern, err)
+		return MatchesWildcard(pattern, str)
+	}
+	return matcher.Match(str)
+}
+
+// CompileWildcardPattern converts a wildcard pattern to a compiled regex.
 func CompileWildcardPattern(pattern string) (*regexp.Regexp, error) {
 	return compileWildcardPattern(pattern)
 }
 
-// matchWildcardString uses efficient string manipulation for * and ? wildcards
-// This implementation uses a backtracking algorithm that handles both wildcard types
-// without requiring regex compilation.
-//
-// Rules:
-//   - '*' matches any sequence of characters (including empty string)
-//   - '?' matches exactly one character (any character)
 func matchWildcardString(pattern, str string) bool {
-	// Handle simple cases
 	if pattern == "*" {
 		return true
 	}
@@ -179,74 +150,37 @@ func matchWildcardString(pattern, str string) bool {
 
 	targetIndex := 0
 	patternIndex := 0
-
-	// Index of the most recent '*' in the pattern (-1 if none)
 	lastStarIndex := -1
-
-	// Index in target where the last '*' started matching
 	lastStarMatchIndex := 0
 
 	for targetIndex < len(str) {
 		switch {
-		// Case 1: Current characters match directly or '?' matches any single character
-		case patternIndex < len(pattern) &&
-			(pattern[patternIndex] == '?' || pattern[patternIndex] == str[targetIndex]):
-
+		case patternIndex < len(pattern) && (pattern[patternIndex] == '?' || pattern[patternIndex] == str[targetIndex]):
 			targetIndex++
 			patternIndex++
-
-		// Case 2: Wildcard '*' found in pattern
-		case patternIndex < len(pattern) &&
-			pattern[patternIndex] == '*':
-
+		case patternIndex < len(pattern) && pattern[patternIndex] == '*':
 			lastStarIndex = patternIndex
 			lastStarMatchIndex = targetIndex
 			patternIndex++
-
-		// Case 3: Previous '*' can absorb one more character
 		case lastStarIndex != -1:
-
 			patternIndex = lastStarIndex + 1
 			lastStarMatchIndex++
 			targetIndex = lastStarMatchIndex
-
-		// Case 4: No match possible
 		default:
 			return false
 		}
 	}
 
-	// Consume any trailing '*' in the pattern
 	for patternIndex < len(pattern) && pattern[patternIndex] == '*' {
 		patternIndex++
 	}
-
-	// Match is valid only if the entire pattern is consumed
 	return patternIndex == len(pattern)
 }
 
-// matchWildcardRegex uses WildcardMatcher for patterns with ? wildcards
-func matchWildcardRegex(pattern, str string) bool {
-	matcher, err := GetCachedWildcardMatcher(pattern)
-	if err != nil {
-		glog.Errorf("Error getting WildcardMatcher for pattern %s: %v. Falling back to matchWildcardString.", pattern, err)
-		// Fallback to matchWildcardString
-		return matchWildcardString(pattern, str)
-	}
-	return matcher.Match(str)
-}
-
-// compileWildcardPattern converts a wildcard pattern to regex
 func compileWildcardPattern(pattern string) (*regexp.Regexp, error) {
-	// Escape special regex characters except * and ?
 	escaped := regexp.QuoteMeta(pattern)
-
-	// Replace escaped wildcards with regex equivalents
 	escaped = strings.ReplaceAll(escaped, `\*`, `.*`)
 	escaped = strings.ReplaceAll(escaped, `\?`, `.`)
-
-	// Anchor the pattern
 	escaped = "^" + escaped + "$"
-
 	return regexp.Compile(escaped)
 }

weed/util/wildcard/wildcard_matcher_test.go

@@ -0,0 +1,211 @@
+package wildcard
+
+import "testing"
+
+func TestMatchesWildcard(t *testing.T) {
+	tests := []struct {
+		name     string
+		pattern  string
+		str      string
+		expected bool
+	}{
+		{"Exact match", "test", "test", true},
+		{"Single wildcard", "*", "anything", true},
+		{"Empty string with wildcard", "*", "", true},
+		{"Prefix wildcard", "test*", "test123", true},
+		{"Suffix wildcard", "*test", "123test", true},
+		{"Middle wildcard", "test*123", "testABC123", true},
+		{"Multiple wildcards", "test*abc*123", "testXYZabcDEF123", true},
+		{"No match", "test*", "other", false},
+		{"Single question mark", "test?", "test1", true},
+		{"Multiple question marks", "test??", "test12", true},
+		{"Question mark no match", "test?", "test12", false},
+		{"Mixed wildcards", "test*abc?def", "testXYZabc1def", true},
+		{"Empty pattern", "", "", true},
+		{"Empty pattern with string", "", "test", false},
+		{"Pattern with string empty", "test", "", false},
+		{"Pattern with regex special chars", "test[abc]", "test[abc]", true},
+		{"Pattern with dots", "test.txt", "test.txt", true},
+		{"Pattern with dots and wildcard", "*.txt", "test.txt", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := MatchesWildcard(tt.pattern, tt.str)
+			if got != tt.expected {
+				t.Errorf("MatchesWildcard(%q, %q) = %v, want %v", tt.pattern, tt.str, got, tt.expected)
+			}
+		})
+	}
+}
+
+func TestWildcardMatcherMatch(t *testing.T) {
+	tests := []struct {
+		name     string
+		pattern  string
+		inputs   []string
+		expected []bool
+	}{
+		{"Simple star", "test*", []string{"test", "test123", "testing", "other"}, []bool{true, true, true, false}},
+		{"Question mark", "test?", []string{"test1", "test2", "test", "test12"}, []bool{true, true, false, false}},
+		{"Extension filter", "*.txt", []string{"file.txt", "test.txt", "file.doc", "txt"}, []bool{true, true, false, false}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m, err := NewWildcardMatcher(tt.pattern)
+			if err != nil {
+				t.Fatalf("NewWildcardMatcher: %v", err)
+			}
+			for i, s := range tt.inputs {
+				got := m.Match(s)
+				if got != tt.expected[i] {
+					t.Errorf("Match(%q) = %v, want %v", s, got, tt.expected[i])
+				}
+			}
+		})
+	}
+}
+
+func TestCompileWildcardPattern(t *testing.T) {
+	tests := []struct {
+		pattern string
+		input   string
+		want    bool
+	}{
+		{"s3:Get*", "s3:GetObject", true},
+		{"s3:Get?bject", "s3:GetObject", true},
+		{"s3:*Object*", "s3:GetObjectAcl", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.pattern, func(t *testing.T) {
+			re, err := CompileWildcardPattern(tt.pattern)
+			if err != nil {
+				t.Fatalf("CompileWildcardPattern: %v", err)
+			}
+			if got := re.MatchString(tt.input); got != tt.want {
+				t.Errorf("got %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestFastMatchesWildcard(t *testing.T) {
+	tests := []struct {
+		pattern string
+		input   string
+		want    bool
+	}{
+		{"s3:Get*", "s3:GetObject", true},
+		{"s3:Put*", "s3:GetObject", false},
+		{"arn:aws:s3:::bucket/*", "arn:aws:s3:::bucket/file.txt", true},
+		{"user:admin-*", "user:admin-john", true},
+		{"user:admin-*", "user:guest-john", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.pattern+"_"+tt.input, func(t *testing.T) {
+			got := FastMatchesWildcard(tt.pattern, tt.input)
+			if got != tt.want {
+				t.Errorf("FastMatchesWildcard(%q, %q) = %v, want %v", tt.pattern, tt.input, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestWildcardMatcherCaching(t *testing.T) {
+	m1, err := GetCachedWildcardMatcher("s3:Get*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	m2, err := GetCachedWildcardMatcher("s3:Get*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if m1 != m2 {
+		t.Error("expected same cached instance")
+	}
+	if !m1.Match("s3:GetObject") {
+		t.Error("expected match")
+	}
+}
+
+func TestWildcardMatcherCacheBounding(t *testing.T) {
+	wildcardMatcherCache.ClearCache()
+	orig := wildcardMatcherCache.maxSize
+	wildcardMatcherCache.maxSize = 3
+	defer func() {
+		wildcardMatcherCache.maxSize = orig
+		wildcardMatcherCache.ClearCache()
+	}()
+	for _, p := range []string{"p1", "p2", "p3"} {
+		GetCachedWildcardMatcher(p)
+	}
+	size, maxSize := wildcardMatcherCache.GetCacheStats()
+	if size != 3 {
+		t.Errorf("expected size 3, got %d", size)
+	}
+	if maxSize != 3 {
+		t.Errorf("expected maxSize 3, got %d", maxSize)
+	}
+	GetCachedWildcardMatcher("p4")
+	size, _ = wildcardMatcherCache.GetCacheStats()
+	if size != 3 {
+		t.Errorf("expected size 3 after eviction, got %d", size)
+	}
+	wildcardMatcherCache.mu.RLock()
+	defer wildcardMatcherCache.mu.RUnlock()
+	if _, ok := wildcardMatcherCache.matchers["p1"]; ok {
+		t.Error("p1 should have been evicted (LRU)")
+	}
+	if _, ok := wildcardMatcherCache.matchers["p4"]; !ok {
+		t.Error("p4 should be in cache")
+	}
+}
+
+func TestWildcardMatcherCacheLRU(t *testing.T) {
+	wildcardMatcherCache.ClearCache()
+	orig := wildcardMatcherCache.maxSize
+	wildcardMatcherCache.maxSize = 3
+	defer func() {
+		wildcardMatcherCache.maxSize = orig
+		wildcardMatcherCache.ClearCache()
+	}()
+	for _, p := range []string{"p1", "p2", "p3"} {
+		GetCachedWildcardMatcher(p)
+	}
+	GetCachedWildcardMatcher("p1") // access p1 to make it most-recently used
+	GetCachedWildcardMatcher("p4") // should evict p2 (now LRU)
+	wildcardMatcherCache.mu.RLock()
+	defer wildcardMatcherCache.mu.RUnlock()
+	if _, ok := wildcardMatcherCache.matchers["p2"]; ok {
+		t.Error("p2 should be evicted (least recently used)")
+	}
+	if _, ok := wildcardMatcherCache.matchers["p1"]; !ok {
+		t.Error("p1 should remain (recently accessed)")
+	}
+	if _, ok := wildcardMatcherCache.matchers["p3"]; !ok {
+		t.Error("p3 should remain")
+	}
+	if _, ok := wildcardMatcherCache.matchers["p4"]; !ok {
+		t.Error("p4 should be in cache")
+	}
+}
+
+func TestWildcardMatcherCacheClear(t *testing.T) {
+	GetCachedWildcardMatcher("test")
+	wildcardMatcherCache.ClearCache()
+	size, _ := wildcardMatcherCache.GetCacheStats()
+	if size != 0 {
+		t.Errorf("expected 0 after clear, got %d", size)
+	}
+}
+
+func BenchmarkMatchesWildcard(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		MatchesWildcard("s3:Get*", "s3:GetObject")
+	}
+}
+
+func BenchmarkFastMatchesWildcard(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		FastMatchesWildcard("s3:Get*", "s3:GetObject")
+	}
+}

weed/worker/tasks/erasure_coding/detection.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
@@ -12,6 +11,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/placement"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/base"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/util"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types"
@@ -51,7 +51,7 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste
 
 	var planner *ecPlacementPlanner
 
-	allowedCollections := ParseCollectionFilter(ecConfig.CollectionFilter)
+	allowedCollections := wildcard.CompileWildcardMatchers(ecConfig.CollectionFilter)
 
 	// Group metrics by VolumeID to handle replicas and select canonical server
 	volumeGroups := make(map[uint32][]*types.VolumeHealthMetrics)
@@ -108,12 +108,9 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste
 		}
 
 		// Check collection filter if specified
-		if len(allowedCollections) > 0 {
-			// Skip if volume's collection is not in the allowed list
-			if !allowedCollections[metric.Collection] {
-				skippedCollectionFilter++
-				continue
-			}
+		if len(allowedCollections) > 0 && !wildcard.MatchesAnyWildcard(allowedCollections, metric.Collection) {
+			skippedCollectionFilter++
+			continue
 		}
 
 		// Check quiet duration and fullness criteria
@@ -336,20 +333,6 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste
 	return results, hasMore, nil
 }
 
-func ParseCollectionFilter(filter string) map[string]bool {
-	allowed := make(map[string]bool)
-	for _, collection := range strings.Split(filter, ",") {
-		trimmed := strings.TrimSpace(collection)
-		if trimmed != "" {
-			allowed[trimmed] = true
-		}
-	}
-	if len(allowed) == 0 {
-		return nil
-	}
-	return allowed
-}
-
 type ecDiskState struct {
 	baseAvailable      int64
 	reservedVolumes    int32

weed/worker/tasks/erasure_coding/ec_task.go

@@ -34,11 +34,12 @@ type ErasureCodingTask struct {
 	grpcDialOption grpc.DialOption
 
 	// EC parameters
-	dataShards      int32
-	parityShards    int32
-	targets         []*worker_pb.TaskTarget // Unified targets for EC shards
-	sources         []*worker_pb.TaskSource // Unified sources for cleanup
-	shardAssignment map[string][]string     // destination -> assigned shard types
+	dataShards        int32
+	parityShards      int32
+	targets           []*worker_pb.TaskTarget // Unified targets for EC shards
+	sources           []*worker_pb.TaskSource // Unified sources for cleanup
+	shardAssignment   map[string][]string     // destination -> assigned shard types
+	allowSourceDelete bool
 }
 
 // NewErasureCodingTask creates a new unified EC task instance
@@ -106,9 +107,10 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	}
 
 	// Use the working directory from task parameters, or fall back to a default
-	baseWorkDir := t.workDir
-
-	// Create unique working directory for this task
+	baseWorkDir := ecParams.WorkingDir
+	if baseWorkDir == "" {
+		baseWorkDir = t.GetWorkingDir()
+	}
 	taskWorkDir := filepath.Join(baseWorkDir, fmt.Sprintf("vol_%d_%d", t.volumeID, time.Now().Unix()))
 	if err := os.MkdirAll(taskWorkDir, 0755); err != nil {
 		return fmt.Errorf("failed to create task working directory %s: %v", taskWorkDir, err)
@@ -119,9 +121,9 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	t.workDir = taskWorkDir
 	glog.V(1).Infof("Task working directory configured: %s (logs will be written here)", taskWorkDir)
 
-	// Ensure cleanup of working directory (but preserve logs)
+	// Ensure cleanup of working directory
 	defer func() {
-		// Clean up volume files and EC shards, but preserve the directory structure and any logs
+		// Clean up volume files and EC shards
 		patterns := []string{"*.dat", "*.idx", "*.ec*", "*.vif"}
 		for _, pattern := range patterns {
 			matches, err := filepath.Glob(filepath.Join(taskWorkDir, pattern))
@@ -134,7 +136,12 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 				}
 			}
 		}
-		glog.V(1).Infof("Cleaned up volume files from working directory: %s (logs preserved)", taskWorkDir)
+		// Remove the entire working directory
+		if err := os.RemoveAll(taskWorkDir); err != nil {
+			glog.V(2).Infof("Could not remove working directory %s: %v", taskWorkDir, err)
+		} else {
+			glog.V(1).Infof("Cleaned up working directory: %s", taskWorkDir)
+		}
 	}()
 
 	// Step 1: Mark volume readonly
@@ -173,6 +180,7 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	if err := t.mountEcShards(); err != nil {
 		return fmt.Errorf("failed to mount EC shards: %v", err)
 	}
+	t.allowSourceDelete = true
 
 	// Step 6: Delete original volume
 	t.ReportProgressWithStage(90.0, "Deleting original volume")
@@ -407,6 +415,16 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string
 		}).Info("EC index file generated")
 	}
 
+	ecjFile := baseName + ".ecj"
+	if info, err := os.Stat(ecjFile); err == nil {
+		shardFiles["ecj"] = ecjFile
+		t.GetLogger().WithFields(map[string]interface{}{
+			"file_type":  "ecj",
+			"file_path":  ecjFile,
+			"size_bytes": info.Size(),
+		}).Info("EC journal file generated")
+	}
+
 	// Generate .vif file (volume info)
 	vifFile := baseName + ".vif"
 	volumeInfo := &volume_server_pb.VolumeInfo{
@@ -438,266 +456,25 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string
 // distributeEcShards distributes locally generated EC shards to destination servers
 // using pre-assigned shard IDs from planning phase
 func (t *ErasureCodingTask) distributeEcShards(shardFiles map[string]string) error {
-	if len(t.targets) == 0 {
-		return fmt.Errorf("no targets specified for EC shard distribution")
+	assignment, err := erasure_coding.DistributeEcShards(t.volumeID, t.collection, t.targets, shardFiles, t.grpcDialOption, t.GetLogger())
+	if err != nil {
+		return err
 	}
-
-	if len(shardFiles) == 0 {
-		return fmt.Errorf("no shard files available for distribution")
-	}
-
-	// Build shard assignment from pre-assigned target shard IDs (from planning phase)
-	shardAssignment := make(map[string][]string)
-
-	for _, target := range t.targets {
-		if len(target.ShardIds) == 0 {
-			continue // Skip targets with no assigned shards
-		}
-
-		var assignedShards []string
-
-		// Convert shard IDs to shard file names (e.g., 0 → "ec00", 1 → "ec01")
-		for _, shardId := range target.ShardIds {
-			shardType := fmt.Sprintf("ec%02d", shardId)
-			assignedShards = append(assignedShards, shardType)
-		}
-
-		// Add metadata files (.ecx, .vif) to targets that have shards
-		if len(assignedShards) > 0 {
-			if _, hasEcx := shardFiles["ecx"]; hasEcx {
-				assignedShards = append(assignedShards, "ecx")
-			}
-			if _, hasVif := shardFiles["vif"]; hasVif {
-				assignedShards = append(assignedShards, "vif")
-			}
-		}
-
-		shardAssignment[target.Node] = assignedShards
-	}
-
-	if len(shardAssignment) == 0 {
-		return fmt.Errorf("no shard assignments found from planning phase")
-	}
-
-	// Store assignment for use during mounting
-	t.shardAssignment = shardAssignment
-
-	// Send assigned shards to each destination
-	for destNode, assignedShards := range shardAssignment {
-		t.GetLogger().WithFields(map[string]interface{}{
-			"destination":     destNode,
-			"assigned_shards": len(assignedShards),
-			"shard_types":     assignedShards,
-		}).Info("Starting shard distribution to destination server")
-
-		// Send only the assigned shards to this destination
-		var transferredBytes int64
-		for _, shardType := range assignedShards {
-			filePath, exists := shardFiles[shardType]
-			if !exists {
-				return fmt.Errorf("shard file %s not found for destination %s", shardType, destNode)
-			}
-
-			// Log file size before transfer
-			if info, err := os.Stat(filePath); err == nil {
-				transferredBytes += info.Size()
-				t.GetLogger().WithFields(map[string]interface{}{
-					"destination": destNode,
-					"shard_type":  shardType,
-					"file_path":   filePath,
-					"size_bytes":  info.Size(),
-					"size_kb":     float64(info.Size()) / 1024,
-				}).Info("Starting shard file transfer")
-			}
-
-			if err := t.sendShardFileToDestination(destNode, filePath, shardType); err != nil {
-				return fmt.Errorf("failed to send %s to %s: %v", shardType, destNode, err)
-			}
-
-			t.GetLogger().WithFields(map[string]interface{}{
-				"destination": destNode,
-				"shard_type":  shardType,
-			}).Info("Shard file transfer completed")
-		}
-
-		// Log summary for this destination
-		t.GetLogger().WithFields(map[string]interface{}{
-			"destination":        destNode,
-			"shards_transferred": len(assignedShards),
-			"total_bytes":        transferredBytes,
-			"total_mb":           float64(transferredBytes) / (1024 * 1024),
-		}).Info("All shards distributed to destination server")
-	}
-
-	glog.V(1).Infof("Successfully distributed EC shards to %d destinations", len(shardAssignment))
+	t.shardAssignment = assignment
 	return nil
 }
 
-// sendShardFileToDestination sends a single shard file to a destination server using ReceiveFile API
-func (t *ErasureCodingTask) sendShardFileToDestination(destServer, filePath, shardType string) error {
-	return operation.WithVolumeServerClient(false, pb.ServerAddress(destServer), t.grpcDialOption,
-		func(client volume_server_pb.VolumeServerClient) error {
-			// Open the local shard file
-			file, err := os.Open(filePath)
-			if err != nil {
-				return fmt.Errorf("failed to open shard file %s: %v", filePath, err)
-			}
-			defer file.Close()
-
-			// Get file size
-			fileInfo, err := file.Stat()
-			if err != nil {
-				return fmt.Errorf("failed to get file info for %s: %v", filePath, err)
-			}
-
-			// Determine file extension and shard ID
-			var ext string
-			var shardId uint32
-			if shardType == "ecx" {
-				ext = ".ecx"
-				shardId = 0 // ecx file doesn't have a specific shard ID
-			} else if shardType == "vif" {
-				ext = ".vif"
-				shardId = 0 // vif file doesn't have a specific shard ID
-			} else if strings.HasPrefix(shardType, "ec") && len(shardType) == 4 {
-				// EC shard file like "ec00", "ec01", etc.
-				ext = "." + shardType
-				fmt.Sscanf(shardType[2:], "%d", &shardId)
-			} else {
-				return fmt.Errorf("unknown shard type: %s", shardType)
-			}
-
-			// Create streaming client
-			stream, err := client.ReceiveFile(context.Background())
-			if err != nil {
-				return fmt.Errorf("failed to create receive stream: %v", err)
-			}
-
-			// Send file info first
-			err = stream.Send(&volume_server_pb.ReceiveFileRequest{
-				Data: &volume_server_pb.ReceiveFileRequest_Info{
-					Info: &volume_server_pb.ReceiveFileInfo{
-						VolumeId:   t.volumeID,
-						Ext:        ext,
-						Collection: t.collection,
-						IsEcVolume: true,
-						ShardId:    shardId,
-						FileSize:   uint64(fileInfo.Size()),
-					},
-				},
-			})
-			if err != nil {
-				return fmt.Errorf("failed to send file info: %v", err)
-			}
-
-			// Send file content in chunks
-			buffer := make([]byte, 64*1024) // 64KB chunks
-			for {
-				n, readErr := file.Read(buffer)
-				if n > 0 {
-					err = stream.Send(&volume_server_pb.ReceiveFileRequest{
-						Data: &volume_server_pb.ReceiveFileRequest_FileContent{
-							FileContent: buffer[:n],
-						},
-					})
-					if err != nil {
-						return fmt.Errorf("failed to send file content: %v", err)
-					}
-				}
-				if readErr == io.EOF {
-					break
-				}
-				if readErr != nil {
-					return fmt.Errorf("failed to read file: %v", readErr)
-				}
-			}
-
-			// Close stream and get response
-			resp, err := stream.CloseAndRecv()
-			if err != nil {
-				return fmt.Errorf("failed to close stream: %v", err)
-			}
-
-			if resp.Error != "" {
-				return fmt.Errorf("server error: %s", resp.Error)
-			}
-
-			glog.V(2).Infof("Successfully sent %s (%d bytes) to %s", shardType, resp.BytesWritten, destServer)
-			return nil
-		})
-}
-
 // mountEcShards mounts EC shards on destination servers
 func (t *ErasureCodingTask) mountEcShards() error {
-	if t.shardAssignment == nil {
-		return fmt.Errorf("shard assignment not available for mounting")
-	}
-
-	// Mount only assigned shards on each destination
-	for destNode, assignedShards := range t.shardAssignment {
-		// Convert shard names to shard IDs for mounting
-		var shardIds []uint32
-		var metadataFiles []string
-
-		for _, shardType := range assignedShards {
-			// Skip metadata files (.ecx, .vif) - only mount EC shards
-			if strings.HasPrefix(shardType, "ec") && len(shardType) == 4 {
-				// Parse shard ID from "ec00", "ec01", etc.
-				var shardId uint32
-				if _, err := fmt.Sscanf(shardType[2:], "%d", &shardId); err == nil {
-					shardIds = append(shardIds, shardId)
-				}
-			} else {
-				metadataFiles = append(metadataFiles, shardType)
-			}
-		}
-
-		t.GetLogger().WithFields(map[string]interface{}{
-			"destination":    destNode,
-			"shard_ids":      shardIds,
-			"shard_count":    len(shardIds),
-			"metadata_files": metadataFiles,
-		}).Info("Starting EC shard mount operation")
-
-		if len(shardIds) == 0 {
-			t.GetLogger().WithFields(map[string]interface{}{
-				"destination":    destNode,
-				"metadata_files": metadataFiles,
-			}).Info("No EC shards to mount (only metadata files)")
-			continue
-		}
-
-		err := operation.WithVolumeServerClient(false, pb.ServerAddress(destNode), t.grpcDialOption,
-			func(client volume_server_pb.VolumeServerClient) error {
-				_, mountErr := client.VolumeEcShardsMount(context.Background(), &volume_server_pb.VolumeEcShardsMountRequest{
-					VolumeId:   t.volumeID,
-					Collection: t.collection,
-					ShardIds:   shardIds,
-				})
-				return mountErr
-			})
-
-		if err != nil {
-			t.GetLogger().WithFields(map[string]interface{}{
-				"destination": destNode,
-				"shard_ids":   shardIds,
-				"error":       err.Error(),
-			}).Error("Failed to mount EC shards")
-		} else {
-			t.GetLogger().WithFields(map[string]interface{}{
-				"destination": destNode,
-				"shard_ids":   shardIds,
-				"volume_id":   t.volumeID,
-				"collection":  t.collection,
-			}).Info("Successfully mounted EC shards")
-		}
-	}
-
-	return nil
+	return erasure_coding.MountEcShards(t.volumeID, t.collection, t.shardAssignment, t.grpcDialOption, t.GetLogger())
 }
 
 // deleteOriginalVolume deletes the original volume and all its replicas from all servers
 func (t *ErasureCodingTask) deleteOriginalVolume() error {
+	if !t.allowSourceDelete {
+		return fmt.Errorf("source deletion disabled: EC shards not fully distributed and mounted")
+	}
+
 	// Get replicas from task parameters (set during detection)
 	replicas := t.getReplicas()
 

weed/worker/tasks/vacuum/detection.go

@@ -98,10 +98,10 @@ func Detection(metrics []*types.VolumeHealthMetrics, clusterInfo *types.ClusterI
 // This function is moved from MaintenanceIntegration.createVacuumTaskParams to the detection logic
 func createVacuumTaskParams(task *types.TaskDetectionResult, metric *types.VolumeHealthMetrics, vacuumConfig *Config, clusterInfo *types.ClusterInfo) *worker_pb.TaskParams {
 	// Use configured values or defaults
-	garbageThreshold := 0.3                    // Default 30%
-	verifyChecksum := true                     // Default to verify
-	batchSize := int32(1000)                   // Default batch size
-	workingDir := "/tmp/seaweedfs_vacuum_work" // Default working directory
+	garbageThreshold := 0.3  // Default 30%
+	verifyChecksum := true   // Default to verify
+	batchSize := int32(1000) // Default batch size
+	workingDir := ""         // Use worker-provided default if empty
 
 	if vacuumConfig != nil {
 		garbageThreshold = vacuumConfig.GarbageThreshold

weed/worker/types/base/task.go

@@ -16,6 +16,7 @@ type BaseTask struct {
 	logger           types.Logger
 	cancelled        bool
 	currentStage     string
+	workingDir       string
 }
 
 // NewBaseTask creates a new base task
@@ -116,3 +117,13 @@ func (t *BaseTask) EstimateTime(params *worker_pb.TaskParams) time.Duration {
 	// Subclasses must implement this
 	return 0
 }
+
+// SetWorkingDir sets the task working directory
+func (t *BaseTask) SetWorkingDir(workingDir string) {
+	t.workingDir = workingDir
+}
+
+// GetWorkingDir returns the task working directory
+func (t *BaseTask) GetWorkingDir() string {
+	return t.workingDir
+}

weed/worker/types/task.go

@@ -29,6 +29,10 @@ type Task interface {
 	// Progress
 	GetProgress() float64
 	SetProgressCallback(func(float64, string))
+
+	// Working Directory
+	SetWorkingDir(string)
+	GetWorkingDir() string
 }
 
 // TaskWithLogging extends Task with logging capabilities
@@ -131,6 +135,7 @@ type UnifiedBaseTask struct {
 	logger           Logger
 	cancelled        bool
 	currentStage     string
+	workingDir       string
 }
 
 // NewBaseTask creates a new base task
@@ -206,3 +211,13 @@ func (t *UnifiedBaseTask) SetLogger(logger Logger) {
 func (t *UnifiedBaseTask) GetLogger() Logger {
 	return t.logger
 }
+
+// SetWorkingDir sets the task working directory
+func (t *UnifiedBaseTask) SetWorkingDir(workingDir string) {
+	t.workingDir = workingDir
+}
+
+// GetWorkingDir returns the task working directory
+func (t *UnifiedBaseTask) GetWorkingDir() string {
+	return t.workingDir
+}

weed/worker/worker.go

@@ -172,6 +172,13 @@ func NewWorker(config *types.WorkerConfig) (*Worker, error) {
 	// Use the global unified registry that already has all tasks registered
 	registry := tasks.GetGlobalTaskRegistry()
 
+	// Ensure the base working directory exists
+	if config.BaseWorkingDir != "" {
+		if err := os.MkdirAll(config.BaseWorkingDir, 0755); err != nil {
+			return nil, fmt.Errorf("failed to create base working directory %s: %v", config.BaseWorkingDir, err)
+		}
+	}
+
 	// Initialize task log handler
 	logDir := filepath.Join(config.BaseWorkingDir, "task_logs")
 	// Ensure the base task log directory exists to avoid errors when admin requests logs
@@ -677,6 +684,7 @@ func (w *Worker) executeTask(task *types.TaskInput) {
 
 	// Task execution uses the new unified Task interface
 	glog.V(2).Infof("Executing task %s in working directory: %s", task.ID, taskWorkingDir)
+	taskInstance.SetWorkingDir(taskWorkingDir)
 
 	// If we have a file logger, adapt it so task WithFields logs are captured into file
 	if fileLogger != nil {