rename flag from -showSystemEntries to -includeSystemEntries

* plugin scheduler: run iceberg and lifecycle lanes concurrently

The default lane serialises job types under a single admin lock
because volume-management operations share global state. Iceberg
and lifecycle lanes have no such constraint, so run each of their
job types independently in separate goroutines.

* Fix concurrent lane scheduler status

* plugin scheduler: address review feedback

- Extract collectDueJobTypes helper to deduplicate policy loading
  between locked and concurrent iteration paths.
- Use atomic.Bool instead of sync.Mutex for hadJobs in the concurrent
  path.
- Set lane loop state to "busy" before launching concurrent goroutines
  so the lane is not reported as idle while work runs.
- Convert TestLaneRequiresLock to table-driven style.
- Add TestRunLaneSchedulerIterationLockBehavior to verify the scheduler
  acquires the admin lock only for lanes that require it.
- Fix flaky TestGetLaneSchedulerStatusShowsActiveConcurrentLaneWork by
  not starting background scheduler goroutines that race with the
  direct runJobTypeIteration call.

.github/workflows/container_latest.yml

@@ -26,6 +26,7 @@ on:
 
 permissions:
   contents: read
+  security-events: write
 
 jobs:
   setup:
@@ -149,9 +150,48 @@ jobs:
           # Remove Go build cache
           sudo rm -rf /tmp/go-build*
 
-  create-manifest:
+  trivy-scan:
     runs-on: ubuntu-latest
     needs: [setup, build]
+    strategy:
+      matrix:
+        variant: ${{ fromJSON(needs.setup.outputs.variants) }}
+    steps:
+      - name: Configure variant
+        id: config
+        run: |
+          if [ "${{ matrix.variant }}" == "large_disk" ]; then
+            echo "tag_suffix=_large_disk" >> $GITHUB_OUTPUT
+          else
+            echo "tag_suffix=" >> $GITHUB_OUTPUT
+          fi
+      - name: Login to GHCR
+        uses: docker/login-action@v4
+        with:
+          registry: ghcr.io
+          username: ${{ secrets.GHCR_USERNAME }}
+          password: ${{ secrets.GHCR_TOKEN }}
+      - name: Run Trivy vulnerability scanner
+        # Pin to SHA — mutable tags were compromised (GHSA-69fq-xp46-6x23)
+        uses: aquasecurity/trivy-action@57a97c7e7821a5776cebc9bb87c984fa69cba8f1 # v0.35.0
+        with:
+          # Scan amd64 only — OS packages are identical across architectures
+          # since they all use the same alpine base, so a single-arch scan
+          # provides sufficient coverage without multiplying CI time.
+          image-ref: ghcr.io/chrislusf/seaweedfs:${{ github.event_name == 'workflow_dispatch' && github.event.inputs.image_tag || 'latest' }}${{ steps.config.outputs.tag_suffix }}-amd64
+          format: sarif
+          output: trivy-results.sarif
+          severity: HIGH,CRITICAL
+          exit-code: '1'
+      - name: Upload Trivy scan results to GitHub Security
+        uses: github/codeql-action/upload-sarif@v3
+        if: always()
+        with:
+          sarif_file: trivy-results.sarif
+
+  create-manifest:
+    runs-on: ubuntu-latest
+    needs: [setup, build, trivy-scan]
     if: github.event_name != 'pull_request'
     strategy:
       matrix:

.github/workflows/s3-mutation-regression-tests.yml

@@ -0,0 +1,133 @@
+name: "S3 Mutation Regression Tests"
+
+on:
+  pull_request:
+    paths:
+      - 'weed/s3api/**'
+      - 'test/s3/delete/**'
+      - 'test/s3/distributed_lock/**'
+      - 'test/s3/versioning/**'
+      - 'test/volume_server/framework/**'
+      - 'docker/compose/s3.json'
+      - '.github/workflows/s3-mutation-regression-tests.yml'
+
+concurrency:
+  group: ${{ github.head_ref }}/s3-mutation-regression-tests
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  s3-versioning-regressions:
+    name: S3 Versioning Regression Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 25
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+
+      - name: Run S3 versioning regression tests
+        timeout-minutes: 20
+        working-directory: test/s3/versioning
+        run: |
+          set -x
+          make test-with-server TEST_PATTERN="TestVersioningCompleteMultipartUploadIsIdempotent|TestVersioningSelfCopyMetadataReplaceCreatesNewVersion|TestVersioningSelfCopyMetadataReplaceSuspendedKeepsNullVersion|TestSuspendedDeleteCreatesDeleteMarker"
+
+      - name: Show server logs on failure
+        if: failure()
+        working-directory: test/s3/versioning
+        run: |
+          echo "=== Server Logs ==="
+          if [ -f weed-test.log ]; then
+            tail -100 weed-test.log
+          fi
+
+      - name: Upload versioning logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: s3-versioning-regression-logs
+          path: test/s3/versioning/weed-test*.log
+          retention-days: 3
+
+  s3-delete-regressions:
+    name: S3 Delete Regression Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 20
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+
+      - name: Run S3 delete regression tests
+        timeout-minutes: 15
+        working-directory: test/s3/delete
+        run: |
+          set -x
+          make test-with-server
+
+      - name: Show server logs on failure
+        if: failure()
+        working-directory: test/s3/delete
+        run: |
+          echo "=== Server Logs ==="
+          if [ -f weed-test.log ]; then
+            tail -100 weed-test.log
+          fi
+
+      - name: Upload delete logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: s3-delete-regression-logs
+          path: test/s3/delete/weed-test*.log
+          retention-days: 3
+
+  s3-distributed-lock-regressions:
+    name: S3 Distributed Lock Regression Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 30
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+
+      - name: Build SeaweedFS
+        run: |
+          go build -o weed/weed -buildvcs=false ./weed
+
+      - name: Run distributed lock regressions
+        timeout-minutes: 25
+        env:
+          TMPDIR: ${{ github.workspace }}/test/s3/distributed_lock/tmp
+          S3_DISTRIBUTED_LOCK_KEEP_LOGS: "1"
+          WEED_BINARY: ${{ github.workspace }}/weed/weed
+        run: |
+          set -x
+          mkdir -p "$TMPDIR"
+          go test -v -count=1 -timeout=20m ./test/s3/distributed_lock
+
+      - name: Upload distributed lock logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: s3-distributed-lock-regression-logs
+          path: test/s3/distributed_lock/tmp/seaweedfs_s3_distributed_lock_*
+          retention-days: 3

docker/Dockerfile.go_build

@@ -79,5 +79,9 @@ RUN mkdir -p /data/filerldb2 && \
 VOLUME /data
 WORKDIR /data
 
-# Entrypoint will handle permission fixes and user switching
+# Run as non-root by default (satisfies security scanners).
+# Use `docker run --user root` if you need the entrypoint to fix
+# /data volume ownership before dropping privileges.
+USER seaweed
+
 ENTRYPOINT ["/entrypoint.sh"]

docker/Dockerfile.local

@@ -37,5 +37,9 @@ RUN mkdir -p /data/filerldb2 && \
 VOLUME /data
 WORKDIR /data
 
-# Entrypoint will handle permission fixes and user switching
+# Run as non-root by default (satisfies security scanners).
+# Use `docker run --user root` if you need the entrypoint to fix
+# /data volume ownership before dropping privileges.
+USER seaweed
+
 ENTRYPOINT ["/entrypoint.sh"]

test/fuse_integration/git_operations_test.go

@@ -431,6 +431,17 @@ func resetToCommitWithRecovery(t *testing.T, bareRepo, localClone, mountClone, c
 			}
 			continue
 		}
+		// Verify git commands actually work in the directory — the kernel
+		// dcache can transiently show the dir then lose it after reset.
+		if _, err := tryGitCommand(mountClone, "rev-parse", "HEAD"); err != nil {
+			lastErr = fmt.Errorf("post-reset verification failed: %w", err)
+			if attempt < maxAttempts {
+				t.Logf("reset recovery attempt %d: %v — removing clone for re-create", attempt, lastErr)
+				os.RemoveAll(mountClone)
+				time.Sleep(2 * time.Second)
+			}
+			continue
+		}
 		return
 	}
 	require.NoError(t, lastErr, "git reset --hard %s failed after %d recovery attempts", commit, maxAttempts)

test/s3/delete/Makefile

@@ -0,0 +1,98 @@
+# S3 delete regression test harness
+
+.PHONY: all test help build-weed check-deps start-server stop-server test-delete test-with-server clean logs
+
+WEED_BINARY := ../../../weed/weed_binary
+S3_PORT := 8333
+TEST_TIMEOUT := 10m
+SERVER_DIR := ./test-volume-data
+S3_CONFIG := ./test_s3.json
+
+.DEFAULT_GOAL := help
+
+all: test
+
+test: test-with-server
+
+help:
+	@echo "S3 delete regression test harness"
+	@echo ""
+	@echo "Available targets:"
+	@echo "  build-weed      - Build the SeaweedFS binary"
+	@echo "  start-server    - Start SeaweedFS with S3 enabled"
+	@echo "  stop-server     - Stop the test server"
+	@echo "  test-delete     - Run delete regression tests"
+	@echo "  test-with-server - Start server, run tests, stop server"
+	@echo "  logs            - Show server logs"
+	@echo "  clean           - Remove test artifacts"
+
+build-weed:
+	@echo "Building SeaweedFS binary..."
+	@cd ../../../weed && go build -o weed_binary .
+	@chmod +x $(WEED_BINARY)
+
+check-deps: build-weed
+	@command -v go >/dev/null 2>&1 || (echo "Go is required but not installed" && exit 1)
+	@test -f $(WEED_BINARY) || (echo "SeaweedFS binary not found at $(WEED_BINARY)" && exit 1)
+	@test -f $(S3_CONFIG) || (echo "S3 config not found at $(S3_CONFIG)" && exit 1)
+
+start-server: check-deps
+	@echo "Starting SeaweedFS server for delete regression tests..."
+	@rm -f weed-server.pid
+	@mkdir -p $(SERVER_DIR)
+	@$(WEED_BINARY) mini \
+		-dir=$(SERVER_DIR) \
+		-s3 \
+		-s3.port=$(S3_PORT) \
+		-s3.config=$(S3_CONFIG) \
+		-master.peers=none \
+		> weed-test.log 2>&1 & echo $$! > weed-server.pid
+	@for i in $$(seq 1 60); do \
+		if curl -sS -o /dev/null http://127.0.0.1:$(S3_PORT)/ >/dev/null 2>&1; then \
+			echo "SeaweedFS S3 server is ready on port $(S3_PORT)"; \
+			exit 0; \
+		fi; \
+		sleep 1; \
+	done; \
+	echo "SeaweedFS S3 server failed to start"; \
+	echo "=== Server logs ==="; \
+	test -f weed-test.log && cat weed-test.log || true; \
+	exit 1
+
+stop-server:
+	@echo "Stopping SeaweedFS server..."
+	@if [ -f weed-server.pid ]; then \
+		SERVER_PID=$$(cat weed-server.pid); \
+		kill -TERM $$SERVER_PID 2>/dev/null || true; \
+		for i in $$(seq 1 20); do \
+			if ! kill -0 $$SERVER_PID >/dev/null 2>&1; then \
+				break; \
+			fi; \
+			sleep 0.1; \
+		done; \
+		if kill -0 $$SERVER_PID >/dev/null 2>&1; then \
+			kill -KILL $$SERVER_PID 2>/dev/null || true; \
+		fi; \
+		rm -f weed-server.pid; \
+	fi
+
+test-delete: check-deps
+	@echo "Running delete regression tests..."
+	@go test -v -timeout=$(TEST_TIMEOUT) .
+
+test-with-server: start-server
+	@trap "$(MAKE) stop-server" EXIT; \
+	$(MAKE) test-delete || (echo "=== Last 100 lines of server logs ===" && tail -100 weed-test.log && exit 1)
+	@$(MAKE) stop-server
+
+logs:
+	@if test -f weed-test.log; then \
+		tail -f weed-test.log; \
+	else \
+		echo "No server log file found."; \
+	fi
+
+clean:
+	@$(MAKE) stop-server
+	@rm -f weed-test.log weed-server.pid
+	@rm -rf $(SERVER_DIR)

test/s3/delete/test_s3.json

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

test/s3/distributed_lock/distributed_lock_test.go

@@ -74,33 +74,31 @@ func runConditionalPutRace(t *testing.T, clients []s3RaceClient, bucket, key str
 	t.Helper()
 
 	start := make(chan struct{})
-	results := make(chan putAttemptResult, len(clients)*2)
+	results := make(chan putAttemptResult, len(clients))
 	var wg sync.WaitGroup
 
 	for _, client := range clients {
-		for attempt := 0; attempt < 2; attempt++ {
-			wg.Add(1)
-			body := fmt.Sprintf("%s-attempt-%d", client.name, attempt)
-			go func(client s3RaceClient, body string) {
-				defer wg.Done()
-				<-start
+		wg.Add(1)
+		body := fmt.Sprintf("%s-race", client.name)
+		go func(client s3RaceClient, body string) {
+			defer wg.Done()
+			<-start
 
-				ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
-				defer cancel()
+			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+			defer cancel()
 
-				_, err := client.client.PutObject(ctx, &s3.PutObjectInput{
-					Bucket:      aws.String(bucket),
-					Key:         aws.String(key),
-					IfNoneMatch: aws.String("*"),
-					Body:        bytes.NewReader([]byte(body)),
-				})
-				results <- putAttemptResult{
-					clientName: client.name,
-					body:       body,
-					err:        err,
-				}
-			}(client, body)
-		}
+			_, err := client.client.PutObject(ctx, &s3.PutObjectInput{
+				Bucket:      aws.String(bucket),
+				Key:         aws.String(key),
+				IfNoneMatch: aws.String("*"),
+				Body:        bytes.NewReader([]byte(body)),
+			})
+			results <- putAttemptResult{
+				clientName: client.name,
+				body:       body,
+				err:        err,
+			}
+		}(client, body)
 	}
 
 	close(start)
@@ -127,7 +125,7 @@ func runConditionalPutRace(t *testing.T, clients []s3RaceClient, bucket, key str
 
 	require.Empty(t, unexpectedErrors, "unexpected race errors")
 	require.Equal(t, 1, successes, "exactly one write should win")
-	require.Equal(t, len(clients)*2-1, preconditionFailures, "all losing writes should fail with 412")
+	require.Equal(t, len(clients)-1, preconditionFailures, "all losing writes should fail with 412")
 
 	object, err := clients[0].client.GetObject(context.Background(), &s3.GetObjectInput{
 		Bucket: aws.String(bucket),

test/s3/versioning/Makefile

@@ -1,7 +1,7 @@
 # S3 API Test Makefile
 # This Makefile provides comprehensive targets for running S3 versioning tests
 
-.PHONY: help build-weed setup-server start-server stop-server test-versioning test-versioning-quick test-versioning-comprehensive test-all clean logs check-deps
+.PHONY: all test help build-weed setup-server start-server stop-server test-versioning test-versioning-quick test-versioning-regressions test-versioning-comprehensive test-all clean logs check-deps
 
 # Configuration
 WEED_BINARY := ../../../weed/weed_binary
@@ -12,6 +12,12 @@ FILER_PORT := 8888
 TEST_TIMEOUT := 10m
 TEST_PATTERN := TestVersioning
 
+.DEFAULT_GOAL := help
+
+all: test
+
+test: test-with-server
+
 # Default target
 help:
 	@echo "S3 API Test Makefile"
@@ -25,6 +31,7 @@ help:
 	@echo "  stop-server            - Stop SeaweedFS server"
 	@echo "  test-versioning        - Run all versioning tests"
 	@echo "  test-versioning-quick  - Run core versioning tests only"
+	@echo "  test-versioning-regressions - Run targeted versioning regression tests"
 	@echo "  test-versioning-simple - Run tests without server management"
 	@echo "  test-versioning-comprehensive - Run comprehensive versioning tests"
 	@echo "  test-all               - Run all S3 API tests"
@@ -179,6 +186,12 @@ test-versioning-quick: check-deps
 	@go test -v -timeout=$(TEST_TIMEOUT) -run "TestBucketListReturnDataVersioning|TestVersioningBasicWorkflow|TestVersioningDeleteMarkers" .
 	@echo "✅ Core versioning tests completed"
 
+# Targeted regressions for conditional mutation, multipart, copy, and suspended delete flows.
+test-versioning-regressions: check-deps
+	@echo "Running targeted S3 versioning regression tests..."
+	@go test -v -timeout=$(TEST_TIMEOUT) -run "TestVersioningCompleteMultipartUploadIsIdempotent|TestVersioningSelfCopyMetadataReplaceCreatesNewVersion|TestVersioningSelfCopyMetadataReplaceSuspendedKeepsNullVersion|TestSuspendedDeleteCreatesDeleteMarker" .
+	@echo "✅ Targeted versioning regression tests completed"
+
 # All versioning tests
 test-versioning: check-deps
 	@echo "Running all S3 versioning tests..."
@@ -348,4 +361,4 @@ compare-python-tests:
 	@echo "Python equivalent: test_bucket_list_return_data_versioning"
 	@echo ""
 	@echo "Running Go version..."
-	@time go test -v -run "TestBucketListReturnDataVersioning" . 2>&1 | grep -E "(PASS|FAIL|took)" 
+	@time go test -v -run "TestBucketListReturnDataVersioning" . 2>&1 | grep -E "(PASS|FAIL|took)" 

weed/admin/plugin/plugin_scheduler.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -106,6 +107,14 @@ func (r *Plugin) schedulerLoop() {
 
 // runLaneSchedulerIteration runs one scheduling pass for a single lane,
 // processing only the job types assigned to that lane.
+//
+// For lanes that require a lock (e.g. LaneDefault), all job types are
+// processed sequentially under one admin lock because their volume
+// management operations share global state.
+//
+// For lanes that do not require a lock (e.g. LaneIceberg, LaneLifecycle),
+// each job type runs independently in its own goroutine so they do not
+// block each other.
 func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool {
 	r.expireStaleJobs(time.Now().UTC())
 
@@ -122,21 +131,23 @@ func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool {
 		return false
 	}
 
-	r.setLaneLoopState(ls, "", "waiting_for_lock")
-	lockName := fmt.Sprintf("plugin scheduler:%s", ls.lane)
-	releaseLock, err := r.acquireAdminLock(lockName)
-	if err != nil {
-		glog.Warningf("Plugin scheduler [%s] failed to acquire lock: %v", ls.lane, err)
-		r.setLaneLoopState(ls, "", "idle")
-		return false
-	}
-	if releaseLock != nil {
-		defer releaseLock()
+	if LaneRequiresLock(ls.lane) {
+		return r.runLaneSchedulerIterationLocked(ls, jobTypes)
 	}
+	return r.runLaneSchedulerIterationConcurrent(ls, jobTypes)
+}
 
-	active := make(map[string]struct{}, len(jobTypes))
-	hadJobs := false
+// dueJobType pairs a job type with its resolved scheduling policy.
+type dueJobType struct {
+	jobType string
+	policy  schedulerPolicy
+}
 
+// collectDueJobTypes loads policies for all job types in the lane and
+// returns those whose detection interval has elapsed. It also returns
+// the full set of active job type names for later pruning.
+func (r *Plugin) collectDueJobTypes(ls *schedulerLaneState, jobTypes []string) (active map[string]struct{}, due []dueJobType) {
+	active = make(map[string]struct{}, len(jobTypes))
 	for _, jobType := range jobTypes {
 		active[jobType] = struct{}{}
 
@@ -156,9 +167,31 @@ func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool {
 		if !r.markDetectionDue(jobType, policy.DetectionInterval, initialDelay) {
 			continue
 		}
+		due = append(due, dueJobType{jobType: jobType, policy: policy})
+	}
+	return active, due
+}
 
-		detected := r.runJobTypeIteration(jobType, policy)
-		if detected {
+// runLaneSchedulerIterationLocked processes job types sequentially under a
+// single admin lock. Used by the default lane where volume management
+// operations must be serialised.
+func (r *Plugin) runLaneSchedulerIterationLocked(ls *schedulerLaneState, jobTypes []string) bool {
+	r.setLaneLoopState(ls, "", "waiting_for_lock")
+	lockName := fmt.Sprintf("plugin scheduler:%s", ls.lane)
+	releaseLock, err := r.acquireAdminLock(lockName)
+	if err != nil {
+		glog.Warningf("Plugin scheduler [%s] failed to acquire lock: %v", ls.lane, err)
+		r.setLaneLoopState(ls, "", "idle")
+		return false
+	}
+	if releaseLock != nil {
+		defer releaseLock()
+	}
+
+	active, due := r.collectDueJobTypes(ls, jobTypes)
+	hadJobs := false
+	for _, w := range due {
+		if r.runJobTypeIteration(w.jobType, w.policy) {
 			hadJobs = true
 		}
 	}
@@ -169,6 +202,33 @@ func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool {
 	return hadJobs
 }
 
+// runLaneSchedulerIterationConcurrent processes each job type in its own
+// goroutine so they run independently. Used by lanes (e.g. iceberg,
+// lifecycle) whose job types do not share global state.
+func (r *Plugin) runLaneSchedulerIterationConcurrent(ls *schedulerLaneState, jobTypes []string) bool {
+	active, due := r.collectDueJobTypes(ls, jobTypes)
+
+	r.setLaneLoopState(ls, "", "busy")
+
+	var hadJobs atomic.Bool
+	var wg sync.WaitGroup
+	for _, w := range due {
+		wg.Add(1)
+		go func(jobType string, policy schedulerPolicy) {
+			defer wg.Done()
+			if r.runJobTypeIteration(jobType, policy) {
+				hadJobs.Store(true)
+			}
+		}(w.jobType, w.policy)
+	}
+	wg.Wait()
+
+	r.pruneSchedulerState(active)
+	r.pruneDetectorLeases(active)
+	r.setLaneLoopState(ls, "", "idle")
+	return hadJobs.Load()
+}
+
 // runSchedulerIteration is kept for backward compatibility. It runs a
 // single iteration across ALL job types (equivalent to the old single-loop
 // behavior). It is only used by the legacy schedulerLoop() fallback.
@@ -267,7 +327,7 @@ func (r *Plugin) wakeScheduler() {
 func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) bool {
 	r.recordSchedulerRunStart(jobType)
 	r.clearWaitingJobQueue(jobType)
-	r.setSchedulerLoopState(jobType, "detecting")
+	r.setSchedulerLoopStateForJobType(jobType, "detecting")
 	r.markJobTypeInFlight(jobType)
 	defer r.finishDetection(jobType)
 
@@ -399,7 +459,7 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 		return detected
 	}
 
-	r.setSchedulerLoopState(jobType, "executing")
+	r.setSchedulerLoopStateForJobType(jobType, "executing")
 
 	// Scan proposals for the maximum estimated_runtime_seconds so the
 	// execution phase gets enough time for large jobs (e.g. vacuum on

weed/admin/plugin/plugin_scheduler_test.go

@@ -3,6 +3,7 @@ package plugin
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -594,3 +595,81 @@ func TestPickDetectorReassignsWhenLeaseIsStale(t *testing.T) {
 		t.Fatalf("expected detector lease to be updated to worker-a, got=%s", lease)
 	}
 }
+
+// trackingLockManager records whether Acquire was called and how many times.
+type trackingLockManager struct {
+	mu       sync.Mutex
+	acquired int
+}
+
+func (m *trackingLockManager) Acquire(reason string) (func(), error) {
+	m.mu.Lock()
+	m.acquired++
+	m.mu.Unlock()
+	return func() {}, nil
+}
+
+func (m *trackingLockManager) count() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.acquired
+}
+
+func TestRunLaneSchedulerIterationLockBehavior(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		lane       SchedulerLane
+		jobType    string
+		wantLock   bool
+	}{
+		{"Default", LaneDefault, "vacuum", true},
+		{"Iceberg", LaneIceberg, "iceberg_maintenance", false},
+		{"Lifecycle", LaneLifecycle, "s3_lifecycle", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			lm := &trackingLockManager{}
+			pluginSvc, err := New(Options{
+				LockManager: lm,
+				ClusterContextProvider: func(context.Context) (*plugin_pb.ClusterContext, error) {
+					return &plugin_pb.ClusterContext{}, nil
+				},
+			})
+			if err != nil {
+				t.Fatalf("New: %v", err)
+			}
+			defer pluginSvc.Shutdown()
+
+			// Register a detectable worker for the job type.
+			pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+				WorkerId: "worker-a",
+				Capabilities: []*plugin_pb.JobTypeCapability{
+					{JobType: tt.jobType, CanDetect: true},
+				},
+			})
+
+			// Enable the job type so the scheduler picks it up.
+			err = pluginSvc.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{
+				JobType: tt.jobType,
+				AdminRuntime: &plugin_pb.AdminRuntimeConfig{
+					Enabled:                  true,
+					DetectionIntervalSeconds: 1,
+				},
+			})
+			if err != nil {
+				t.Fatalf("SaveJobTypeConfig: %v", err)
+			}
+
+			ls := pluginSvc.lanes[tt.lane]
+			pluginSvc.runLaneSchedulerIteration(ls)
+
+			if got := lm.count(); (got > 0) != tt.wantLock {
+				t.Errorf("lock acquired %d times, wantLock=%v", got, tt.wantLock)
+			}
+		})
+	}
+}

weed/admin/plugin/scheduler_lane.go

@@ -38,6 +38,25 @@ var laneIdleSleep = map[SchedulerLane]time.Duration{
 	LaneLifecycle: 5 * time.Minute,
 }
 
+// laneRequiresLock maps each lane to whether its job types must be
+// serialised under a single admin lock. The default lane needs this
+// because volume-management operations share global state. Other
+// lanes run each job type independently.
+var laneRequiresLock = map[SchedulerLane]bool{
+	LaneDefault:   true,
+	LaneIceberg:   false,
+	LaneLifecycle: false,
+}
+
+// LaneRequiresLock returns true if the given lane needs a single admin
+// lock to serialise its job types. Unknown lanes default to true.
+func LaneRequiresLock(lane SchedulerLane) bool {
+	if v, ok := laneRequiresLock[lane]; ok {
+		return v
+	}
+	return true
+}
+
 // LaneIdleSleep returns the idle sleep duration for the given lane,
 // falling back to defaultSchedulerIdleSleep if the lane is unknown.
 func LaneIdleSleep(lane SchedulerLane) time.Duration {

weed/admin/plugin/scheduler_lane_test.go

@@ -27,6 +27,26 @@ func TestAllLanesHaveIdleSleep(t *testing.T) {
 	}
 }
 
+func TestLaneRequiresLock(t *testing.T) {
+	tests := []struct {
+		name string
+		lane SchedulerLane
+		want bool
+	}{
+		{"Default", LaneDefault, true},
+		{"Iceberg", LaneIceberg, false},
+		{"Lifecycle", LaneLifecycle, false},
+		{"Unknown", "unknown_lane", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := LaneRequiresLock(tt.lane); got != tt.want {
+				t.Errorf("LaneRequiresLock(%q) = %v, want %v", tt.lane, got, tt.want)
+			}
+		})
+	}
+}
+
 func TestKnownJobTypesInMap(t *testing.T) {
 	// Ensure the well-known job types are mapped. This catches drift
 	// if a handler's job type string changes without updating the map.

weed/admin/plugin/scheduler_status.go

@@ -195,6 +195,21 @@ func (r *Plugin) setSchedulerLoopState(jobType, phase string) {
 	r.schedulerLoopMu.Unlock()
 }
 
+// setSchedulerLoopStateForJobType keeps the aggregate scheduler state and the
+// owning lane state in sync while a specific job type is active.
+func (r *Plugin) setSchedulerLoopStateForJobType(jobType, phase string) {
+	if r == nil {
+		return
+	}
+	r.setSchedulerLoopState(jobType, phase)
+	if jobType == "" {
+		return
+	}
+	if ls := r.lanes[JobTypeLane(jobType)]; ls != nil {
+		r.setLaneLoopState(ls, jobType, phase)
+	}
+}
+
 func (r *Plugin) recordSchedulerIterationComplete(hadJobs bool) {
 	if r == nil {
 		return
@@ -251,7 +266,6 @@ func (r *Plugin) aggregateLaneLoopStates() schedulerLoopState {
 	return agg
 }
 
-
 // --- Per-lane loop state helpers ---
 
 func (r *Plugin) setLaneLoopState(ls *schedulerLaneState, jobType, phase string) {

weed/admin/plugin/scheduler_status_test.go

@@ -1,7 +1,9 @@
 package plugin
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 )
@@ -62,3 +64,84 @@ func TestGetSchedulerStatusIncludesLastDetectionCount(t *testing.T) {
 		t.Fatalf("expected job type status for %s", jobType)
 	}
 }
+
+func TestGetLaneSchedulerStatusShowsActiveConcurrentLaneWork(t *testing.T) {
+	clusterContextStarted := make(chan struct{})
+	releaseClusterContext := make(chan struct{})
+
+	// Create the Plugin without a ClusterContextProvider so no background
+	// scheduler goroutines are started; they would race with the direct
+	// runJobTypeIteration call below.
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	// Set the provider after construction so runJobTypeIteration can use it.
+	pluginSvc.clusterContextProvider = func(context.Context) (*plugin_pb.ClusterContext, error) {
+		close(clusterContextStarted)
+		<-releaseClusterContext
+		return nil, context.Canceled
+	}
+
+	const jobType = "s3_lifecycle"
+	err = pluginSvc.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{
+		JobType: jobType,
+		AdminRuntime: &plugin_pb.AdminRuntimeConfig{
+			Enabled:                  true,
+			DetectionIntervalSeconds: 30,
+			DetectionTimeoutSeconds:  15,
+		},
+	})
+	if err != nil {
+		t.Fatalf("SaveJobTypeConfig: %v", err)
+	}
+
+	policy, enabled, err := pluginSvc.loadSchedulerPolicy(jobType)
+	if err != nil {
+		t.Fatalf("loadSchedulerPolicy: %v", err)
+	}
+	if !enabled {
+		t.Fatalf("expected enabled policy")
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		pluginSvc.runJobTypeIteration(jobType, policy)
+	}()
+
+	select {
+	case <-clusterContextStarted:
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for job type iteration to start")
+	}
+
+	var laneStatus SchedulerStatus
+	var aggregateStatus SchedulerStatus
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		laneStatus = pluginSvc.GetLaneSchedulerStatus(LaneLifecycle)
+		aggregateStatus = pluginSvc.GetSchedulerStatus()
+		if laneStatus.CurrentJobType == jobType && laneStatus.CurrentPhase == "detecting" &&
+			aggregateStatus.CurrentJobType == jobType && aggregateStatus.CurrentPhase == "detecting" {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	if laneStatus.CurrentJobType != jobType || laneStatus.CurrentPhase != "detecting" {
+		t.Fatalf("unexpected lane status while work is active: job=%q phase=%q", laneStatus.CurrentJobType, laneStatus.CurrentPhase)
+	}
+	if aggregateStatus.CurrentJobType != jobType || aggregateStatus.CurrentPhase != "detecting" {
+		t.Fatalf("unexpected aggregate status while work is active: job=%q phase=%q", aggregateStatus.CurrentJobType, aggregateStatus.CurrentPhase)
+	}
+
+	close(releaseClusterContext)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for job type iteration to finish")
+	}
+}

weed/command/mount.go

@@ -31,6 +31,7 @@ type MountOptions struct {
 	uidMap             *string
 	gidMap             *string
 	readOnly           *bool
+	includeSystemEntries *bool
 	debug              *bool
 	debugPort          *int
 	localSocket        *string
@@ -99,6 +100,7 @@ func init() {
 	mountOptions.uidMap = cmdMount.Flag.String("map.uid", "", "map local uid to uid on filer, comma-separated <local_uid>:<filer_uid>")
 	mountOptions.gidMap = cmdMount.Flag.String("map.gid", "", "map local gid to gid on filer, comma-separated <local_gid>:<filer_gid>")
 	mountOptions.readOnly = cmdMount.Flag.Bool("readOnly", false, "read only")
+	mountOptions.includeSystemEntries = cmdMount.Flag.Bool("includeSystemEntries", false, "show filer system entries (e.g. /topics, /etc) in directory listings")
 	mountOptions.debug = cmdMount.Flag.Bool("debug", false, "serves runtime profiling data, e.g., http://localhost:<debug.port>/debug/pprof/goroutine?debug=2")
 	mountOptions.debugPort = cmdMount.Flag.Int("debug.port", 6061, "http port for debugging")
 	mountOptions.localSocket = cmdMount.Flag.String("localSocket", "", "default to /tmp/seaweedfs-mount-<mount_dir_hash>.sock")

weed/command/mount_std.go

@@ -340,6 +340,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		VolumeServerAccess:          *mountOptions.volumeServerAccess,
 		Cipher:                      cipher,
 		UidGidMapper:                uidGidMapper,
+		IncludeSystemEntries:        *option.includeSystemEntries,
 		DisableXAttr:                *option.disableXAttr,
 		IsMacOs:                     runtime.GOOS == "darwin",
 		MetadataFlushSeconds:        *option.metadataFlushSeconds,

weed/mount/meta_cache/meta_cache.go

@@ -27,18 +27,19 @@ type MetaCache struct {
 	localStore   filer.VirtualFilerStore
 	leveldbStore *leveldb.LevelDBStore // direct reference for batch operations
 	sync.RWMutex
-	uidGidMapper      *UidGidMapper
-	markCachedFn      func(fullpath util.FullPath)
-	isCachedFn        func(fullpath util.FullPath) bool
-	invalidateFunc    func(fullpath util.FullPath, entry *filer_pb.Entry)
-	onDirectoryUpdate func(dir util.FullPath)
-	visitGroup        singleflight.Group // deduplicates concurrent EnsureVisited calls for the same path
-	applyCh           chan metadataApplyRequest
-	applyDone         chan struct{}
-	applyStateMu      sync.Mutex
-	applyClosed       bool
-	buildingDirs      map[util.FullPath]*directoryBuildState
-	dedupRing         dedupRingBuffer
+	uidGidMapper         *UidGidMapper
+	markCachedFn         func(fullpath util.FullPath)
+	isCachedFn           func(fullpath util.FullPath) bool
+	invalidateFunc       func(fullpath util.FullPath, entry *filer_pb.Entry)
+	onDirectoryUpdate    func(dir util.FullPath)
+	visitGroup           singleflight.Group // deduplicates concurrent EnsureVisited calls for the same path
+	applyCh              chan metadataApplyRequest
+	applyDone            chan struct{}
+	applyStateMu         sync.Mutex
+	applyClosed          bool
+	buildingDirs         map[util.FullPath]*directoryBuildState
+	dedupRing            dedupRingBuffer
+	includeSystemEntries bool
 }
 
 var errMetaCacheClosed = errors.New("metadata cache is shut down")
@@ -84,17 +85,18 @@ type metadataApplyRequest struct {
 	done         chan error
 }
 
-func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath,
+func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, includeSystemEntries bool,
 	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(util.FullPath, *filer_pb.Entry), onDirectoryUpdate func(dir util.FullPath)) *MetaCache {
 	leveldbStore, virtualStore := openMetaStore(dbFolder)
 	mc := &MetaCache{
-		root:              root,
-		localStore:        virtualStore,
-		leveldbStore:      leveldbStore,
-		markCachedFn:      markCachedFn,
-		isCachedFn:        isCachedFn,
-		uidGidMapper:      uidGidMapper,
-		onDirectoryUpdate: onDirectoryUpdate,
+		root:                 root,
+		localStore:           virtualStore,
+		leveldbStore:         leveldbStore,
+		markCachedFn:         markCachedFn,
+		isCachedFn:           isCachedFn,
+		uidGidMapper:         uidGidMapper,
+		onDirectoryUpdate:    onDirectoryUpdate,
+		includeSystemEntries: includeSystemEntries,
 		invalidateFunc: func(fullpath util.FullPath, entry *filer_pb.Entry) {
 			invalidateFunc(fullpath, entry)
 		},

weed/mount/meta_cache/meta_cache_apply_test.go

@@ -312,6 +312,7 @@ func newTestMetaCache(t *testing.T, cached map[util.FullPath]bool) (*MetaCache,
 		filepath.Join(t.TempDir(), "meta"),
 		mapper,
 		util.FullPath("/"),
+		false,
 		func(path util.FullPath) {
 			cachedMu.Lock()
 			defer cachedMu.Unlock()

weed/mount/meta_cache/meta_cache_init.go

@@ -107,7 +107,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl
 			var err error
 			snapshotTsNs, err = filer_pb.ReadDirAllEntriesWithSnapshot(ctx, client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
 				entry := filer.FromPbEntry(string(path), pbEntry)
-				if IsHiddenSystemEntry(string(path), entry.Name()) {
+				if !mc.includeSystemEntries && IsHiddenSystemEntry(string(path), entry.Name()) {
 					return nil
 				}
 

weed/mount/weedfs.go

@@ -65,6 +65,7 @@ type Option struct {
 	VolumeServerAccess string // how to access volume servers
 	Cipher             bool   // whether encrypt data on volume server
 	UidGidMapper       *meta_cache.UidGidMapper
+	IncludeSystemEntries bool
 
 	// Periodic metadata flush interval in seconds (0 to disable)
 	// This protects chunks from being purged by volume.fsck for long-running writes
@@ -203,6 +204,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 
 	wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.getUniqueCacheDirForRead(), "meta"), option.UidGidMapper,
 		util.FullPath(option.FilerMountRootPath),
+		option.IncludeSystemEntries,
 		func(path util.FullPath) {
 			wfs.inodeToPath.MarkChildrenCached(path)
 		}, func(path util.FullPath) bool {

weed/mount/weedfs_dir_read.go

@@ -302,7 +302,7 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out *fuse.DirEntryList,
 	if input.Offset >= dh.entryStreamOffset {
 		if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset {
 			skipCount := uint32(input.Offset-dh.entryStreamOffset) + batchSize
-			entries, snapshotTs, err := loadDirectoryEntriesDirect(context.Background(), wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs)
+			entries, snapshotTs, err := loadDirectoryEntriesDirect(context.Background(), wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs, wfs.option.IncludeSystemEntries)
 			if err != nil {
 				glog.Errorf("list filer directory: %v", err)
 				return fuse.EIO
@@ -331,7 +331,7 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out *fuse.DirEntryList,
 			}
 		}
 
-		entries, snapshotTs, err := loadDirectoryEntriesDirect(context.Background(), wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs)
+		entries, snapshotTs, err := loadDirectoryEntriesDirect(context.Background(), wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs, wfs.option.IncludeSystemEntries)
 		if err != nil {
 			glog.Errorf("list filer directory: %v", err)
 			return fuse.EIO
@@ -360,13 +360,13 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out *fuse.DirEntryList,
 	return fuse.OK
 }
 
-func loadDirectoryEntriesDirect(ctx context.Context, client filer_pb.FilerClient, uidGidMapper *meta_cache.UidGidMapper, dirPath util.FullPath, startFileName string, includeStart bool, limit uint32, snapshotTsNs int64) ([]*filer.Entry, int64, error) {
+func loadDirectoryEntriesDirect(ctx context.Context, client filer_pb.FilerClient, uidGidMapper *meta_cache.UidGidMapper, dirPath util.FullPath, startFileName string, includeStart bool, limit uint32, snapshotTsNs int64, includeSystemEntries bool) ([]*filer.Entry, int64, error) {
 	entries := make([]*filer.Entry, 0, limit)
 	var actualSnapshotTsNs int64
 	err := client.WithFilerClient(false, func(sc filer_pb.SeaweedFilerClient) error {
 		var innerErr error
 		actualSnapshotTsNs, innerErr = filer_pb.DoSeaweedListWithSnapshot(ctx, sc, dirPath, "", func(entry *filer_pb.Entry, isLast bool) error {
-			if meta_cache.IsHiddenSystemEntry(string(dirPath), entry.Name) {
+			if !includeSystemEntries && meta_cache.IsHiddenSystemEntry(string(dirPath), entry.Name) {
 				return nil
 			}
 			if uidGidMapper != nil && entry.Attributes != nil {

weed/mount/weedfs_dir_read_test.go

@@ -84,7 +84,7 @@ func TestLoadDirectoryEntriesDirectFiltersHiddenEntriesAndMapsIds(t *testing.T)
 		},
 	}
 
-	entries, _, err := loadDirectoryEntriesDirect(context.Background(), client, mapper, util.FullPath("/"), "", false, 10, 0)
+	entries, _, err := loadDirectoryEntriesDirect(context.Background(), client, mapper, util.FullPath("/"), "", false, 10, 0, false)
 	if err != nil {
 		t.Fatalf("loadDirectoryEntriesDirect: %v", err)
 	}
@@ -98,3 +98,25 @@ func TestLoadDirectoryEntriesDirectFiltersHiddenEntriesAndMapsIds(t *testing.T)
 		t.Fatalf("mapped uid/gid = %d/%d, want 10/20", entries[0].Attr.Uid, entries[0].Attr.Gid)
 	}
 }
+
+func TestLoadDirectoryEntriesDirectShowsSystemEntriesWhenEnabled(t *testing.T) {
+	client := &directoryFilerAccessor{
+		client: &directoryListClient{
+			responses: []*filer_pb.ListEntriesResponse{
+				{Entry: &filer_pb.Entry{Name: "topics"}},
+				{Entry: &filer_pb.Entry{Name: "visible"}},
+			},
+		},
+	}
+
+	entries, _, err := loadDirectoryEntriesDirect(context.Background(), client, nil, util.FullPath("/"), "", false, 10, 0, true)
+	if err != nil {
+		t.Fatalf("loadDirectoryEntriesDirect: %v", err)
+	}
+	if got := len(entries); got != 2 {
+		t.Fatalf("entry count = %d, want 2", got)
+	}
+	if entries[0].Name() != "topics" || entries[1].Name() != "visible" {
+		t.Fatalf("entry names = %q, %q, want topics, visible", entries[0].Name(), entries[1].Name())
+	}
+}

weed/mount/weedfs_file_copy_range_test.go

@@ -338,6 +338,7 @@ func newCopyRangeTestWFSWithMetaCache(t *testing.T) *WFS {
 		filepath.Join(t.TempDir(), "meta"),
 		uidGidMapper,
 		root,
+		false,
 		func(path util.FullPath) {
 			wfs.inodeToPath.MarkChildrenCached(path)
 		},

weed/mount/weedfs_file_mkrm_test.go

@@ -120,6 +120,7 @@ func newCreateTestWFS(t *testing.T) (*WFS, *createEntryTestServer) {
 		filepath.Join(t.TempDir(), "meta"),
 		uidGidMapper,
 		root,
+		false,
 		func(path util.FullPath) {
 			wfs.inodeToPath.MarkChildrenCached(path)
 		},

weed/mount/weedfs_rename_test.go

@@ -23,6 +23,7 @@ func TestHandleRenameResponseLeavesUncachedTargetOutOfCache(t *testing.T) {
 		filepath.Join(t.TempDir(), "meta"),
 		uidGidMapper,
 		root,
+		false,
 		func(path util.FullPath) {
 			inodeToPath.MarkChildrenCached(path)
 		},

weed/plugin/worker/lifecycle/detection.go

@@ -10,11 +10,15 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
+const lifecycleXMLKey = "s3-bucket-lifecycle-configuration-xml"
+
 // detectBucketsWithLifecycleRules scans all S3 buckets to find those
-// with lifecycle (TTL) rules configured in filer.conf.
+// with lifecycle rules, either TTL entries in filer.conf or lifecycle
+// XML stored in bucket metadata.
 func (h *Handler) detectBucketsWithLifecycleRules(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
@@ -53,25 +57,38 @@ func (h *Handler) detectBucketsWithLifecycleRules(
 			continue
 		}
 
-		// Derive the collection name for this bucket.
+		// Check for lifecycle rules from two sources:
+		// 1. filer.conf TTLs (legacy Expiration.Days fast path)
+		// 2. Stored lifecycle XML in bucket metadata (full rule support)
 		collection := bucketName
 		ttls := fc.GetCollectionTtls(collection)
-		if len(ttls) == 0 {
+
+		hasLifecycleXML := entry.Extended != nil && len(entry.Extended[lifecycleXMLKey]) > 0
+		versioningStatus := ""
+		if entry.Extended != nil {
+			versioningStatus = string(entry.Extended[s3_constants.ExtVersioningKey])
+		}
+
+		ruleCount := int64(len(ttls))
+		if !hasLifecycleXML && ruleCount == 0 {
 			continue
 		}
 
-		glog.V(2).Infof("s3_lifecycle: bucket %s has %d lifecycle rule(s)", bucketName, len(ttls))
+		glog.V(2).Infof("s3_lifecycle: bucket %s has %d TTL rule(s), lifecycle_xml=%v, versioning=%s",
+			bucketName, ruleCount, hasLifecycleXML, versioningStatus)
 
 		proposal := &plugin_pb.JobProposal{
 			ProposalId: fmt.Sprintf("s3_lifecycle:%s", bucketName),
 			JobType:    jobType,
-			Summary:    fmt.Sprintf("Lifecycle management for bucket %s (%d rules)", bucketName, len(ttls)),
+			Summary:    fmt.Sprintf("Lifecycle management for bucket %s", bucketName),
 			DedupeKey:  fmt.Sprintf("s3_lifecycle:%s", bucketName),
 			Parameters: map[string]*plugin_pb.ConfigValue{
-				"bucket":       {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: bucketName}},
-				"buckets_path": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: bucketsPath}},
-				"collection":   {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: collection}},
-				"rule_count":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(len(ttls))}},
+				"bucket":             {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: bucketName}},
+				"buckets_path":       {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: bucketsPath}},
+				"collection":         {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: collection}},
+				"rule_count":         {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: ruleCount}},
+				"has_lifecycle_xml":  {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: hasLifecycleXML}},
+				"versioning_status":  {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: versioningStatus}},
 			},
 			Labels: map[string]string{
 				"bucket": bucketName,

weed/plugin/worker/lifecycle/detection_test.go

@@ -0,0 +1,132 @@
+package lifecycle
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func TestBucketHasLifecycleXML(t *testing.T) {
+	tests := []struct {
+		name     string
+		extended map[string][]byte
+		want     bool
+	}{
+		{
+			name:     "has_lifecycle_xml",
+			extended: map[string][]byte{lifecycleXMLKey: []byte("<LifecycleConfiguration/>")},
+			want:     true,
+		},
+		{
+			name:     "empty_lifecycle_xml",
+			extended: map[string][]byte{lifecycleXMLKey: {}},
+			want:     false,
+		},
+		{
+			name:     "no_lifecycle_xml",
+			extended: map[string][]byte{"other-key": []byte("value")},
+			want:     false,
+		},
+		{
+			name:     "nil_extended",
+			extended: nil,
+			want:     false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.extended != nil && len(tt.extended[lifecycleXMLKey]) > 0
+			if got != tt.want {
+				t.Errorf("hasLifecycleXML = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestBucketVersioningStatus(t *testing.T) {
+	tests := []struct {
+		name     string
+		extended map[string][]byte
+		want     string
+	}{
+		{
+			name: "versioning_enabled",
+			extended: map[string][]byte{
+				s3_constants.ExtVersioningKey: []byte("Enabled"),
+			},
+			want: "Enabled",
+		},
+		{
+			name: "versioning_suspended",
+			extended: map[string][]byte{
+				s3_constants.ExtVersioningKey: []byte("Suspended"),
+			},
+			want: "Suspended",
+		},
+		{
+			name:     "no_versioning",
+			extended: map[string][]byte{},
+			want:     "",
+		},
+		{
+			name:     "nil_extended",
+			extended: nil,
+			want:     "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var got string
+			if tt.extended != nil {
+				got = string(tt.extended[s3_constants.ExtVersioningKey])
+			}
+			if got != tt.want {
+				t.Errorf("versioningStatus = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestDetectionProposalParameters(t *testing.T) {
+	// Verify that bucket entries with lifecycle XML or TTL rules produce
+	// proposals with the expected parameters.
+	t.Run("bucket_with_lifecycle_xml_and_versioning", func(t *testing.T) {
+		entry := &filer_pb.Entry{
+			Name:        "my-bucket",
+			IsDirectory: true,
+			Extended: map[string][]byte{
+				lifecycleXMLKey:              []byte(`<LifecycleConfiguration><Rule><Status>Enabled</Status></Rule></LifecycleConfiguration>`),
+				s3_constants.ExtVersioningKey: []byte("Enabled"),
+			},
+		}
+
+		hasXML := entry.Extended != nil && len(entry.Extended[lifecycleXMLKey]) > 0
+		versioning := ""
+		if entry.Extended != nil {
+			versioning = string(entry.Extended[s3_constants.ExtVersioningKey])
+		}
+
+		if !hasXML {
+			t.Error("expected hasLifecycleXML=true")
+		}
+		if versioning != "Enabled" {
+			t.Errorf("expected versioning=Enabled, got %q", versioning)
+		}
+	})
+
+	t.Run("bucket_without_lifecycle_or_ttl_is_skipped", func(t *testing.T) {
+		entry := &filer_pb.Entry{
+			Name:        "empty-bucket",
+			IsDirectory: true,
+			Extended:    map[string][]byte{},
+		}
+
+		hasXML := entry.Extended != nil && len(entry.Extended[lifecycleXMLKey]) > 0
+		ttlCount := 0 // simulated: no TTL rules in filer.conf
+
+		if hasXML || ttlCount > 0 {
+			t.Error("expected bucket to be skipped (no lifecycle XML, no TTLs)")
+		}
+	})
+}

weed/plugin/worker/lifecycle/execution.go

@@ -2,9 +2,12 @@ package lifecycle
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"path"
+	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -13,8 +16,11 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 	pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 )
 
+var errLimitReached = errors.New("limit reached")
+
 type executionResult struct {
 	objectsExpired     int64
 	objectsScanned     int64
@@ -37,17 +43,37 @@ func (h *Handler) executeLifecycleForBucket(
 ) (*executionResult, error) {
 	result := &executionResult{}
 
-	// Load filer.conf to verify TTL rules still exist.
-	fc, err := loadFilerConf(ctx, filerClient)
-	if err != nil {
-		return result, fmt.Errorf("load filer conf: %w", err)
+	// Try to load lifecycle rules from stored XML first (full rule evaluation).
+	// Fall back to filer.conf TTL-only evaluation only if no XML is configured.
+	// If XML exists but is malformed, fail closed (don't fall back to TTL,
+	// which could apply broader rules and delete objects the XML rules would keep).
+	// Transient filer errors fall back to TTL with a warning.
+	lifecycleRules, xmlErr := loadLifecycleRulesFromBucket(ctx, filerClient, bucketsPath, bucket)
+	if xmlErr != nil && errors.Is(xmlErr, errMalformedLifecycleXML) {
+		glog.Errorf("s3_lifecycle: bucket %s: %v (skipping bucket)", bucket, xmlErr)
+		return result, xmlErr
 	}
+	if xmlErr != nil {
+		glog.V(1).Infof("s3_lifecycle: bucket %s: transient error loading lifecycle XML: %v, falling back to TTL", bucket, xmlErr)
+	}
+	// lifecycleRules is non-nil when XML was present (even if empty/all disabled).
+	// Only fall back to TTL when XML was truly absent (nil).
+	xmlPresent := xmlErr == nil && lifecycleRules != nil
+	useRuleEval := xmlPresent && len(lifecycleRules) > 0
 
-	collection := bucket
-	ttlRules := fc.GetCollectionTtls(collection)
-	if len(ttlRules) == 0 {
-		glog.V(1).Infof("s3_lifecycle: bucket %s has no lifecycle rules, skipping", bucket)
-		return result, nil
+	if !useRuleEval && !xmlPresent {
+		// Fall back to filer.conf TTL rules only when no lifecycle XML exists.
+		// When XML is present but has no effective rules, skip TTL fallback.
+		fc, err := loadFilerConf(ctx, filerClient)
+		if err != nil {
+			return result, fmt.Errorf("load filer conf: %w", err)
+		}
+		collection := bucket
+		ttlRules := fc.GetCollectionTtls(collection)
+		if len(ttlRules) == 0 {
+			glog.V(1).Infof("s3_lifecycle: bucket %s has no lifecycle rules, skipping", bucket)
+			return result, nil
+		}
 	}
 
 	_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
@@ -56,14 +82,23 @@ func (h *Handler) executeLifecycleForBucket(
 		State:           plugin_pb.JobState_JOB_STATE_RUNNING,
 		ProgressPercent: 10,
 		Stage:           "scanning",
-		Message:         fmt.Sprintf("scanning bucket %s for expired objects (%d rules)", bucket, len(ttlRules)),
+		Message:         fmt.Sprintf("scanning bucket %s for expired objects", bucket),
 	})
 
 	// Shared budget across all phases so we don't exceed MaxDeletesPerBucket.
 	remaining := config.MaxDeletesPerBucket
 
-	// Find expired objects.
-	expired, scanned, err := listExpiredObjects(ctx, filerClient, bucketsPath, bucket, remaining)
+	// Find expired objects using rule-based evaluation or TTL fallback.
+	var expired []expiredObject
+	var scanned int64
+	var err error
+	if useRuleEval {
+		expired, scanned, err = listExpiredObjectsByRules(ctx, filerClient, bucketsPath, bucket, lifecycleRules, remaining)
+	} else if !xmlPresent {
+		// TTL-only scan when no lifecycle XML exists.
+		expired, scanned, err = listExpiredObjects(ctx, filerClient, bucketsPath, bucket, remaining)
+	}
+	// When xmlPresent but no effective rules (all disabled), skip object scanning.
 	result.objectsScanned = scanned
 	if err != nil {
 		return result, fmt.Errorf("list expired objects: %w", err)
@@ -140,6 +175,9 @@ func (h *Handler) executeLifecycleForBucket(
 			})
 		}
 
+		// Clean up .versions directories left empty after version deletion.
+		cleanupEmptyVersionsDirectories(ctx, filerClient, expired)
+
 		remaining -= result.objectsExpired + result.errors
 		if remaining < 0 {
 			remaining = 0
@@ -153,7 +191,7 @@ func (h *Handler) executeLifecycleForBucket(
 			State: plugin_pb.JobState_JOB_STATE_RUNNING,
 			Stage: "cleaning_delete_markers", Message: "cleaning expired delete markers",
 		})
-		cleaned, cleanErrs, cleanCtxErr := cleanupDeleteMarkers(ctx, filerClient, bucketsPath, bucket, remaining)
+		cleaned, cleanErrs, cleanCtxErr := cleanupDeleteMarkers(ctx, filerClient, bucketsPath, bucket, lifecycleRules, remaining)
 		result.deleteMarkersClean = int64(cleaned)
 		result.errors += int64(cleanErrs)
 		if cleanCtxErr != nil {
@@ -166,7 +204,23 @@ func (h *Handler) executeLifecycleForBucket(
 	}
 
 	// Abort incomplete multipart uploads.
-	if config.AbortMPUDays > 0 && remaining > 0 {
+	// When lifecycle XML exists, evaluate each upload against the rules
+	// (respecting per-rule prefix filters and DaysAfterInitiation).
+	// Fall back to worker config abort_mpu_days only when no lifecycle
+	// XML is configured for the bucket.
+	if xmlPresent && remaining > 0 {
+		_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
+			JobId: jobID, JobType: jobType,
+			State: plugin_pb.JobState_JOB_STATE_RUNNING,
+			Stage: "aborting_mpus", Message: "evaluating MPU abort rules",
+		})
+		aborted, abortErrs, abortCtxErr := abortMPUsByRules(ctx, filerClient, bucketsPath, bucket, lifecycleRules, remaining)
+		result.mpuAborted = int64(aborted)
+		result.errors += int64(abortErrs)
+		if abortCtxErr != nil {
+			return result, abortCtxErr
+		}
+	} else if !xmlPresent && config.AbortMPUDays > 0 && remaining > 0 {
 		_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
 			JobId: jobID, JobType: jobType,
 			State: plugin_pb.JobState_JOB_STATE_RUNNING,
@@ -183,18 +237,19 @@ func (h *Handler) executeLifecycleForBucket(
 	return result, nil
 }
 
-// cleanupDeleteMarkers scans the bucket for entries marked as delete markers
-// (via the S3 versioning extended attribute) and removes them.
+// cleanupDeleteMarkers scans versioned objects and removes delete markers
+// that are the sole remaining version. This matches AWS S3
+// ExpiredObjectDeleteMarker semantics: a delete marker is only removed when
+// it is the only version of an object (no non-current versions behind it).
 //
-// NOTE: This currently removes delete markers unconditionally without checking
-// whether prior non-expired versions exist. In versioned buckets, removing a
-// delete marker can resurface an older version. A future enhancement should
-// query version metadata before removal to match AWS ExpiredObjectDeleteMarker
-// semantics (only remove when no non-current versions remain).
+// This phase should run AFTER NoncurrentVersionExpiration (PR 4) so that
+// non-current versions have already been cleaned up, potentially leaving
+// delete markers as sole versions eligible for removal.
 func cleanupDeleteMarkers(
 	ctx context.Context,
 	client filer_pb.SeaweedFilerClient,
 	bucketsPath, bucket string,
+	rules []s3lifecycle.Rule,
 	limit int64,
 ) (cleaned, errors int, ctxErr error) {
 	bucketPath := path.Join(bucketsPath, bucket)
@@ -210,14 +265,72 @@ func cleanupDeleteMarkers(
 
 		listErr := filer_pb.SeaweedList(ctx, client, dir, "", func(entry *filer_pb.Entry, isLast bool) error {
 			if entry.IsDirectory {
-				// Skip .uploads directories.
-				if entry.Name != ".uploads" {
-					dirsToProcess = append(dirsToProcess, path.Join(dir, entry.Name))
+				if dir == bucketPath && entry.Name == s3_constants.MultipartUploadsFolder {
+					return nil
 				}
+				if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
+					versionsDir := path.Join(dir, entry.Name)
+					// Check if the latest version is a delete marker.
+					latestIsMarker := string(entry.Extended[s3_constants.ExtLatestVersionIsDeleteMarker]) == "true"
+					if !latestIsMarker {
+						return nil
+					}
+					// Count versions in the directory.
+					versionCount := 0
+					countErr := filer_pb.SeaweedList(ctx, client, versionsDir, "", func(ve *filer_pb.Entry, _ bool) error {
+						if !ve.IsDirectory {
+							versionCount++
+						}
+						return nil
+					}, "", false, 10000)
+					if countErr != nil {
+						glog.V(1).Infof("s3_lifecycle: failed to count versions in %s: %v", versionsDir, countErr)
+						errors++
+						return nil
+					}
+					// Only remove if the delete marker is the sole version.
+					if versionCount != 1 {
+						return nil
+					}
+					// Check that a matching ExpiredObjectDeleteMarker rule exists.
+					// The rule's prefix filter must match this object's key.
+					relDir := strings.TrimPrefix(versionsDir, bucketPath+"/")
+					objKey := strings.TrimSuffix(relDir, s3_constants.VersionsFolder)
+					if len(rules) > 0 && !matchesDeleteMarkerRule(rules, objKey) {
+						return nil
+					}
+					// Find and remove the sole delete marker entry.
+					removedHere := false
+					removeErr := filer_pb.SeaweedList(ctx, client, versionsDir, "", func(ve *filer_pb.Entry, _ bool) error {
+						if !ve.IsDirectory && isDeleteMarker(ve) {
+							if err := filer_pb.DoRemove(ctx, client, versionsDir, ve.Name, true, false, false, false, nil); err != nil {
+								glog.V(1).Infof("s3_lifecycle: failed to remove delete marker %s/%s: %v", versionsDir, ve.Name, err)
+								errors++
+							} else {
+								cleaned++
+								removedHere = true
+							}
+						}
+						return nil
+					}, "", false, 10)
+					if removeErr != nil {
+						glog.V(1).Infof("s3_lifecycle: failed to scan for delete marker in %s: %v", versionsDir, removeErr)
+					}
+					// Remove the now-empty .versions directory only if we
+					// actually deleted the marker in this specific directory.
+					if removedHere {
+						_ = filer_pb.DoRemove(ctx, client, dir, entry.Name, true, true, true, false, nil)
+					}
+					return nil
+				}
+				dirsToProcess = append(dirsToProcess, path.Join(dir, entry.Name))
 				return nil
 			}
 
-			if isDeleteMarker(entry) {
+			// For non-versioned objects: only clean up if explicitly a delete marker
+			// and a matching rule exists.
+			relKey := strings.TrimPrefix(path.Join(dir, entry.Name), bucketPath+"/")
+			if isDeleteMarker(entry) && matchesDeleteMarkerRule(rules, relKey) {
 				if err := filer_pb.DoRemove(ctx, client, dir, entry.Name, true, false, false, false, nil); err != nil {
 					glog.V(1).Infof("s3_lifecycle: failed to remove delete marker %s/%s: %v", dir, entry.Name, err)
 					errors++
@@ -251,6 +364,74 @@ func isDeleteMarker(entry *filer_pb.Entry) bool {
 	return string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true"
 }
 
+// matchesDeleteMarkerRule checks if any enabled ExpiredObjectDeleteMarker rule
+// matches the given object key using the full filter model (prefix, tags, size).
+// When no lifecycle rules are provided (nil means no XML configured),
+// falls back to legacy behavior (returns true to allow cleanup).
+// A non-nil empty slice means XML was present but had no matching rules,
+// so cleanup is not allowed.
+func matchesDeleteMarkerRule(rules []s3lifecycle.Rule, objKey string) bool {
+	if rules == nil {
+		return true // legacy fallback: no lifecycle XML configured
+	}
+	// Delete markers have no size or tags, so build a minimal ObjectInfo.
+	obj := s3lifecycle.ObjectInfo{Key: objKey}
+	for _, r := range rules {
+		if r.Status == "Enabled" && r.ExpiredObjectDeleteMarker && s3lifecycle.MatchesFilter(r, obj) {
+			return true
+		}
+	}
+	return false
+}
+
+// abortMPUsByRules scans the .uploads directory and evaluates each upload
+// against lifecycle rules using EvaluateMPUAbort, which respects per-rule
+// prefix filters and DaysAfterInitiation thresholds.
+func abortMPUsByRules(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	bucketsPath, bucket string,
+	rules []s3lifecycle.Rule,
+	limit int64,
+) (aborted, errs int, ctxErr error) {
+	uploadsDir := path.Join(bucketsPath, bucket, ".uploads")
+	now := time.Now()
+
+	listErr := filer_pb.SeaweedList(ctx, client, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		if !entry.IsDirectory {
+			return nil
+		}
+		if entry.Attributes == nil || entry.Attributes.Crtime <= 0 {
+			return nil
+		}
+
+		createdAt := time.Unix(entry.Attributes.Crtime, 0)
+		result := s3lifecycle.EvaluateMPUAbort(rules, entry.Name, createdAt, now)
+		if result.Action == s3lifecycle.ActionAbortMultipartUpload {
+			uploadPath := path.Join(uploadsDir, entry.Name)
+			if err := filer_pb.DoRemove(ctx, client, uploadsDir, entry.Name, true, true, true, false, nil); err != nil {
+				glog.V(1).Infof("s3_lifecycle: failed to abort MPU %s: %v", uploadPath, err)
+				errs++
+			} else {
+				aborted++
+			}
+		}
+
+		if limit > 0 && int64(aborted+errs) >= limit {
+			return errLimitReached
+		}
+		return nil
+	}, "", false, 10000)
+
+	if listErr != nil && !errors.Is(listErr, errLimitReached) {
+		return aborted, errs, fmt.Errorf("list uploads in %s: %w", uploadsDir, listErr)
+	}
+	return aborted, errs, nil
+}
+
 // abortIncompleteMPUs scans the .uploads directory under a bucket and
 // removes multipart upload entries older than the specified number of days.
 func abortIncompleteMPUs(
@@ -326,3 +507,372 @@ func deleteExpiredObjects(
 func nowUnix() int64 {
 	return time.Now().Unix()
 }
+
+// listExpiredObjectsByRules scans a bucket directory tree and evaluates
+// lifecycle rules against each object using the s3lifecycle evaluator.
+// This function handles non-versioned objects (IsLatest=true). Versioned
+// objects in .versions directories are handled by processVersionsDirectory
+// (added in a separate change for NoncurrentVersionExpiration support).
+func listExpiredObjectsByRules(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	bucketsPath, bucket string,
+	rules []s3lifecycle.Rule,
+	limit int64,
+) ([]expiredObject, int64, error) {
+	var expired []expiredObject
+	var scanned int64
+
+	bucketPath := path.Join(bucketsPath, bucket)
+	now := time.Now()
+	needTags := s3lifecycle.HasTagRules(rules)
+
+	dirsToProcess := []string{bucketPath}
+	for len(dirsToProcess) > 0 {
+		select {
+		case <-ctx.Done():
+			return expired, scanned, ctx.Err()
+		default:
+		}
+
+		dir := dirsToProcess[0]
+		dirsToProcess = dirsToProcess[1:]
+
+		limitReached := false
+		err := filer_pb.SeaweedList(ctx, client, dir, "", func(entry *filer_pb.Entry, isLast bool) error {
+			if entry.IsDirectory {
+				if dir == bucketPath && entry.Name == s3_constants.MultipartUploadsFolder {
+					return nil // skip .uploads at bucket root only
+				}
+				if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
+					versionsDir := path.Join(dir, entry.Name)
+
+					// Evaluate Expiration rules against the latest version.
+					// In versioned buckets, data lives in .versions/ directories,
+					// so we must evaluate the latest version here — it is never
+					// seen as a regular file entry in the parent directory.
+					if obj, ok := latestVersionExpiredByRules(ctx, client, entry, versionsDir, bucketPath, rules, now, needTags); ok {
+						expired = append(expired, obj)
+						scanned++
+						if limit > 0 && int64(len(expired)) >= limit {
+							limitReached = true
+							return errLimitReached
+						}
+					}
+
+					// Process noncurrent versions.
+					vExpired, vScanned, vErr := processVersionsDirectory(ctx, client, versionsDir, bucketPath, rules, now, needTags, limit-int64(len(expired)))
+					if vErr != nil {
+						glog.V(1).Infof("s3_lifecycle: %v", vErr)
+						return vErr
+					}
+					expired = append(expired, vExpired...)
+					scanned += vScanned
+					if limit > 0 && int64(len(expired)) >= limit {
+						limitReached = true
+						return errLimitReached
+					}
+					return nil
+				}
+				dirsToProcess = append(dirsToProcess, path.Join(dir, entry.Name))
+				return nil
+			}
+			scanned++
+
+			// Skip objects already handled by TTL fast path.
+			if entry.Attributes != nil && entry.Attributes.TtlSec > 0 {
+				expirationUnix := entry.Attributes.Crtime + int64(entry.Attributes.TtlSec)
+				if expirationUnix > nowUnix() {
+					return nil // will be expired by RocksDB compaction
+				}
+			}
+
+			// Build ObjectInfo for the evaluator.
+			relKey := strings.TrimPrefix(path.Join(dir, entry.Name), bucketPath+"/")
+			objInfo := s3lifecycle.ObjectInfo{
+				Key:      relKey,
+				IsLatest: true, // non-versioned objects are always "latest"
+			}
+			if entry.Attributes != nil {
+				objInfo.Size = int64(entry.Attributes.GetFileSize())
+				if entry.Attributes.Mtime > 0 {
+					objInfo.ModTime = time.Unix(entry.Attributes.Mtime, 0)
+				} else if entry.Attributes.Crtime > 0 {
+					objInfo.ModTime = time.Unix(entry.Attributes.Crtime, 0)
+				}
+			}
+			if needTags {
+				objInfo.Tags = s3lifecycle.ExtractTags(entry.Extended)
+			}
+
+			result := s3lifecycle.Evaluate(rules, objInfo, now)
+			if result.Action == s3lifecycle.ActionDeleteObject {
+				expired = append(expired, expiredObject{dir: dir, name: entry.Name})
+			}
+
+			if limit > 0 && int64(len(expired)) >= limit {
+				limitReached = true
+				return errLimitReached
+			}
+			return nil
+		}, "", false, 10000)
+
+		if err != nil && !errors.Is(err, errLimitReached) {
+			return expired, scanned, fmt.Errorf("list %s: %w", dir, err)
+		}
+
+		if limitReached || (limit > 0 && int64(len(expired)) >= limit) {
+			break
+		}
+	}
+
+	return expired, scanned, nil
+}
+
+// processVersionsDirectory evaluates NoncurrentVersionExpiration rules
+// against all versions in a .versions directory.
+func processVersionsDirectory(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	versionsDir, bucketPath string,
+	rules []s3lifecycle.Rule,
+	now time.Time,
+	needTags bool,
+	limit int64,
+) ([]expiredObject, int64, error) {
+	var expired []expiredObject
+	var scanned int64
+
+	// Check if any rule has NoncurrentVersionExpiration.
+	hasNoncurrentRules := false
+	for _, r := range rules {
+		if r.Status == "Enabled" && r.NoncurrentVersionExpirationDays > 0 {
+			hasNoncurrentRules = true
+			break
+		}
+	}
+	if !hasNoncurrentRules {
+		return nil, 0, nil
+	}
+
+	// List all versions in this directory.
+	var versions []*filer_pb.Entry
+	listErr := filer_pb.SeaweedList(ctx, client, versionsDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+		if !entry.IsDirectory {
+			versions = append(versions, entry)
+		}
+		return nil
+	}, "", false, 10000)
+	if listErr != nil {
+		return nil, 0, fmt.Errorf("list versions in %s: %w", versionsDir, listErr)
+	}
+	if len(versions) <= 1 {
+		return nil, 0, nil // only one version (the latest), nothing to expire
+	}
+
+	// Sort by version timestamp, newest first.
+	sortVersionsByVersionId(versions)
+
+	// Derive the object key from the .versions directory path.
+	// e.g., /buckets/mybucket/path/to/key.versions -> path/to/key
+	relDir := strings.TrimPrefix(versionsDir, bucketPath+"/")
+	objKey := strings.TrimSuffix(relDir, s3_constants.VersionsFolder)
+
+	// Walk versions: first is latest, rest are non-current.
+	noncurrentIndex := 0
+	for i := 1; i < len(versions); i++ {
+		entry := versions[i]
+		scanned++
+
+		// Skip delete markers from expiration evaluation, but count
+		// them toward NewerNoncurrentVersions so data versions get
+		// the correct noncurrent index.
+		if isDeleteMarker(entry) {
+			noncurrentIndex++
+			continue
+		}
+
+		// Determine successor's timestamp (the version that replaced this one).
+		successorEntry := versions[i-1]
+		successorVersionId := strings.TrimPrefix(successorEntry.Name, "v_")
+		successorTime := s3lifecycle.GetVersionTimestamp(successorVersionId)
+		if successorTime.IsZero() && successorEntry.Attributes != nil && successorEntry.Attributes.Mtime > 0 {
+			successorTime = time.Unix(successorEntry.Attributes.Mtime, 0)
+		}
+
+		objInfo := s3lifecycle.ObjectInfo{
+			Key:              objKey,
+			IsLatest:         false,
+			SuccessorModTime: successorTime,
+			NumVersions:      len(versions),
+			NoncurrentIndex:  noncurrentIndex,
+		}
+		if entry.Attributes != nil {
+			objInfo.Size = int64(entry.Attributes.GetFileSize())
+			if entry.Attributes.Mtime > 0 {
+				objInfo.ModTime = time.Unix(entry.Attributes.Mtime, 0)
+			}
+		}
+		if needTags {
+			objInfo.Tags = s3lifecycle.ExtractTags(entry.Extended)
+		}
+
+		// Evaluate using the detailed ShouldExpireNoncurrentVersion which
+		// handles NewerNoncurrentVersions.
+		for _, rule := range rules {
+			if s3lifecycle.ShouldExpireNoncurrentVersion(rule, objInfo, noncurrentIndex, now) {
+				expired = append(expired, expiredObject{dir: versionsDir, name: entry.Name})
+				break
+			}
+		}
+
+		noncurrentIndex++
+
+		if limit > 0 && int64(len(expired)) >= limit {
+			break
+		}
+	}
+
+	return expired, scanned, nil
+}
+
+// latestVersionExpiredByRules evaluates Expiration rules (Days/Date) against
+// the latest version in a .versions directory. In versioned buckets all data
+// lives inside .versions/ directories, so the latest version is never seen as
+// a regular file entry during the bucket walk. Without this check, Expiration
+// rules would never fire for versioned objects (issue #8757).
+//
+// The .versions directory entry caches metadata about the latest version in
+// its Extended attributes, so we can evaluate expiration without an extra
+// filer round-trip.
+func latestVersionExpiredByRules(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	dirEntry *filer_pb.Entry,
+	versionsDir, bucketPath string,
+	rules []s3lifecycle.Rule,
+	now time.Time,
+	needTags bool,
+) (expiredObject, bool) {
+	if dirEntry.Extended == nil {
+		return expiredObject{}, false
+	}
+
+	// Skip if the latest version is a delete marker — those are handled
+	// by the ExpiredObjectDeleteMarker rule in cleanupDeleteMarkers.
+	if string(dirEntry.Extended[s3_constants.ExtLatestVersionIsDeleteMarker]) == "true" {
+		return expiredObject{}, false
+	}
+
+	latestFileName := string(dirEntry.Extended[s3_constants.ExtLatestVersionFileNameKey])
+	if latestFileName == "" {
+		return expiredObject{}, false
+	}
+
+	// Derive the object key: /buckets/b/path/key.versions → path/key
+	relDir := strings.TrimPrefix(versionsDir, bucketPath+"/")
+	objKey := strings.TrimSuffix(relDir, s3_constants.VersionsFolder)
+
+	objInfo := s3lifecycle.ObjectInfo{
+		Key:      objKey,
+		IsLatest: true,
+	}
+
+	// Populate ModTime from cached metadata.
+	if mtimeStr := string(dirEntry.Extended[s3_constants.ExtLatestVersionMtimeKey]); mtimeStr != "" {
+		if mtime, err := strconv.ParseInt(mtimeStr, 10, 64); err == nil {
+			objInfo.ModTime = time.Unix(mtime, 0)
+		}
+	}
+	if objInfo.ModTime.IsZero() && dirEntry.Attributes != nil && dirEntry.Attributes.Mtime > 0 {
+		objInfo.ModTime = time.Unix(dirEntry.Attributes.Mtime, 0)
+	}
+
+	// Populate Size from cached metadata.
+	if sizeStr := string(dirEntry.Extended[s3_constants.ExtLatestVersionSizeKey]); sizeStr != "" {
+		if size, err := strconv.ParseInt(sizeStr, 10, 64); err == nil {
+			objInfo.Size = size
+		}
+	}
+
+	if needTags {
+		// Tags are stored on the version file entry, not the .versions
+		// directory. Fetch the actual version file to get them.
+		resp, err := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{
+			Directory: versionsDir,
+			Name:      latestFileName,
+		})
+		if err == nil && resp.Entry != nil {
+			objInfo.Tags = s3lifecycle.ExtractTags(resp.Entry.Extended)
+		}
+	}
+
+	result := s3lifecycle.Evaluate(rules, objInfo, now)
+	if result.Action == s3lifecycle.ActionDeleteObject {
+		return expiredObject{dir: versionsDir, name: latestFileName}, true
+	}
+
+	return expiredObject{}, false
+}
+
+// cleanupEmptyVersionsDirectories removes .versions directories that became
+// empty after their contents were deleted. This is called after
+// deleteExpiredObjects to avoid leaving orphaned directories.
+func cleanupEmptyVersionsDirectories(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	deleted []expiredObject,
+) int {
+	// Collect unique .versions directories that had entries deleted.
+	versionsDirs := map[string]struct{}{}
+	for _, obj := range deleted {
+		if strings.HasSuffix(obj.dir, s3_constants.VersionsFolder) {
+			versionsDirs[obj.dir] = struct{}{}
+		}
+	}
+
+	cleaned := 0
+	for vDir := range versionsDirs {
+		if ctx.Err() != nil {
+			break
+		}
+		// Check if the directory is now empty.
+		empty := true
+		listErr := filer_pb.SeaweedList(ctx, client, vDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+			empty = false
+			return errLimitReached // stop after first entry
+		}, "", false, 1)
+
+		if listErr != nil && !errors.Is(listErr, errLimitReached) {
+			glog.V(1).Infof("s3_lifecycle: failed to check if versions dir %s is empty: %v", vDir, listErr)
+			continue
+		}
+
+		if !empty {
+			continue
+		}
+
+		// Remove the empty .versions directory.
+		parentDir, dirName := path.Split(vDir)
+		parentDir = strings.TrimSuffix(parentDir, "/")
+		if err := filer_pb.DoRemove(ctx, client, parentDir, dirName, false, true, true, false, nil); err != nil {
+			glog.V(1).Infof("s3_lifecycle: failed to clean up empty versions dir %s: %v", vDir, err)
+		} else {
+			cleaned++
+		}
+	}
+	return cleaned
+}
+
+// sortVersionsByVersionId sorts version entries newest-first using full
+// version ID comparison (matching compareVersionIds in s3api_version_id.go).
+// This uses the complete version ID string, not just the decoded timestamp,
+// so entries with the same timestamp prefix are correctly ordered by their
+// random suffix.
+func sortVersionsByVersionId(versions []*filer_pb.Entry) {
+	sort.Slice(versions, func(i, j int) bool {
+		vidI := strings.TrimPrefix(versions[i].Name, "v_")
+		vidJ := strings.TrimPrefix(versions[j].Name, "v_")
+		return s3lifecycle.CompareVersionIds(vidI, vidJ) < 0
+	})
+}

weed/plugin/worker/lifecycle/execution_test.go

@@ -0,0 +1,72 @@
+package lifecycle
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+)
+
+func TestMatchesDeleteMarkerRule(t *testing.T) {
+	t.Run("nil_rules_legacy_fallback", func(t *testing.T) {
+		if !matchesDeleteMarkerRule(nil, "any/key") {
+			t.Error("nil rules should return true (legacy fallback)")
+		}
+	})
+
+	t.Run("empty_rules_xml_present_no_match", func(t *testing.T) {
+		rules := []s3lifecycle.Rule{}
+		if matchesDeleteMarkerRule(rules, "any/key") {
+			t.Error("empty rules (XML present) should return false")
+		}
+	})
+
+	t.Run("matching_prefix_rule", func(t *testing.T) {
+		rules := []s3lifecycle.Rule{
+			{ID: "cleanup", Status: "Enabled", Prefix: "logs/", ExpiredObjectDeleteMarker: true},
+		}
+		if !matchesDeleteMarkerRule(rules, "logs/app.log") {
+			t.Error("should match rule with matching prefix")
+		}
+	})
+
+	t.Run("non_matching_prefix", func(t *testing.T) {
+		rules := []s3lifecycle.Rule{
+			{ID: "cleanup", Status: "Enabled", Prefix: "logs/", ExpiredObjectDeleteMarker: true},
+		}
+		if matchesDeleteMarkerRule(rules, "data/file.txt") {
+			t.Error("should not match rule with non-matching prefix")
+		}
+	})
+
+	t.Run("disabled_rule", func(t *testing.T) {
+		rules := []s3lifecycle.Rule{
+			{ID: "cleanup", Status: "Disabled", ExpiredObjectDeleteMarker: true},
+		}
+		if matchesDeleteMarkerRule(rules, "any/key") {
+			t.Error("disabled rule should not match")
+		}
+	})
+
+	t.Run("rule_without_delete_marker_flag", func(t *testing.T) {
+		rules := []s3lifecycle.Rule{
+			{ID: "expire", Status: "Enabled", ExpirationDays: 30},
+		}
+		if matchesDeleteMarkerRule(rules, "any/key") {
+			t.Error("rule without ExpiredObjectDeleteMarker should not match")
+		}
+	})
+
+	t.Run("tag_filtered_rule_no_tags_on_marker", func(t *testing.T) {
+		rules := []s3lifecycle.Rule{
+			{
+				ID: "tagged", Status: "Enabled",
+				ExpiredObjectDeleteMarker: true,
+				FilterTags:               map[string]string{"env": "dev"},
+			},
+		}
+		// Delete markers have no tags, so a tag-filtered rule should not match.
+		if matchesDeleteMarkerRule(rules, "any/key") {
+			t.Error("tag-filtered rule should not match delete marker (no tags)")
+		}
+	})
+}

weed/plugin/worker/lifecycle/integration_test.go

@@ -0,0 +1,781 @@
+package lifecycle
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"net"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+// testFilerServer is an in-memory filer gRPC server for integration tests.
+type testFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	mu      sync.RWMutex
+	entries map[string]*filer_pb.Entry // key: "dir\x00name"
+}
+
+func newTestFilerServer() *testFilerServer {
+	return &testFilerServer{entries: make(map[string]*filer_pb.Entry)}
+}
+
+func (s *testFilerServer) key(dir, name string) string { return dir + "\x00" + name }
+
+func (s *testFilerServer) splitKey(key string) (string, string) {
+	for i := range key {
+		if key[i] == '\x00' {
+			return key[:i], key[i+1:]
+		}
+	}
+	return key, ""
+}
+
+func (s *testFilerServer) putEntry(dir string, entry *filer_pb.Entry) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.entries[s.key(dir, entry.Name)] = proto.Clone(entry).(*filer_pb.Entry)
+}
+
+func (s *testFilerServer) getEntry(dir, name string) *filer_pb.Entry {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	e := s.entries[s.key(dir, name)]
+	if e == nil {
+		return nil
+	}
+	return proto.Clone(e).(*filer_pb.Entry)
+}
+
+func (s *testFilerServer) hasEntry(dir, name string) bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	_, ok := s.entries[s.key(dir, name)]
+	return ok
+}
+
+func (s *testFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	entry, found := s.entries[s.key(req.Directory, req.Name)]
+	if !found {
+		return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}, nil
+}
+
+func (s *testFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error {
+	// Snapshot entries under lock, then stream without holding the lock
+	// (streaming callbacks may trigger DeleteEntry which needs a write lock).
+	s.mu.RLock()
+	var names []string
+	for key := range s.entries {
+		dir, name := s.splitKey(key)
+		if dir == req.Directory {
+			if req.StartFromFileName != "" && name <= req.StartFromFileName {
+				continue
+			}
+			if req.Prefix != "" && !strings.HasPrefix(name, req.Prefix) {
+				continue
+			}
+			names = append(names, name)
+		}
+	}
+	sort.Strings(names)
+
+	// Clone entries while still holding the lock.
+	type namedEntry struct {
+		name  string
+		entry *filer_pb.Entry
+	}
+	snapshot := make([]namedEntry, 0, len(names))
+	for _, name := range names {
+		if req.Limit > 0 && uint32(len(snapshot)) >= req.Limit {
+			break
+		}
+		snapshot = append(snapshot, namedEntry{
+			name:  name,
+			entry: proto.Clone(s.entries[s.key(req.Directory, name)]).(*filer_pb.Entry),
+		})
+	}
+	s.mu.RUnlock()
+
+	// Stream responses without holding any lock.
+	for _, ne := range snapshot {
+		if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: ne.entry}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *testFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.entries[s.key(req.Directory, req.Entry.Name)] = proto.Clone(req.Entry).(*filer_pb.Entry)
+	return &filer_pb.CreateEntryResponse{}, nil
+}
+
+func (s *testFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	k := s.key(req.Directory, req.Name)
+	if _, found := s.entries[k]; !found {
+		return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+	}
+	delete(s.entries, k)
+	if req.IsRecursive {
+		// Delete all descendants: any entry whose directory starts with
+		// the deleted path (handles nested subdirectories).
+		deletedPath := req.Directory + "/" + req.Name
+		for key := range s.entries {
+			dir, _ := s.splitKey(key)
+			if dir == deletedPath || strings.HasPrefix(dir, deletedPath+"/") {
+				delete(s.entries, key)
+			}
+		}
+	}
+	return &filer_pb.DeleteEntryResponse{}, nil
+}
+
+// startTestFiler starts an in-memory filer gRPC server and returns a client.
+func startTestFiler(t *testing.T) (*testFilerServer, filer_pb.SeaweedFilerClient) {
+	t.Helper()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	server := newTestFilerServer()
+	grpcServer := pb.NewGrpcServer()
+	filer_pb.RegisterSeaweedFilerServer(grpcServer, server)
+	go func() { _ = grpcServer.Serve(lis) }()
+
+	t.Cleanup(func() {
+		grpcServer.Stop()
+		_ = lis.Close()
+	})
+
+	host, portStr, err := net.SplitHostPort(lis.Addr().String())
+	if err != nil {
+		t.Fatalf("split host port: %v", err)
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		t.Fatalf("parse port: %v", err)
+	}
+	addr := pb.NewServerAddress(host, 1, port)
+
+	conn, err := pb.GrpcDial(context.Background(), addr.ToGrpcAddress(), false, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+
+	return server, filer_pb.NewSeaweedFilerClient(conn)
+}
+
+// Helper to create a version ID from a timestamp.
+func testVersionId(ts time.Time) string {
+	inverted := math.MaxInt64 - ts.UnixNano()
+	return fmt.Sprintf("%016x", inverted) + "0000000000000000"
+}
+
+func TestIntegration_ListExpiredObjectsByRules(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "test-bucket"
+	bucketDir := bucketsPath + "/" + bucket
+
+	now := time.Now()
+	old := now.Add(-60 * 24 * time.Hour) // 60 days ago
+	recent := now.Add(-5 * 24 * time.Hour) // 5 days ago
+
+	// Create bucket directory.
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+
+	// Create objects.
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name:       "old-file.txt",
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 1024},
+	})
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name:       "recent-file.txt",
+		Attributes: &filer_pb.FuseAttributes{Mtime: recent.Unix(), FileSize: 1024},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "expire-30d", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+
+	expired, scanned, err := listExpiredObjectsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("listExpiredObjectsByRules: %v", err)
+	}
+
+	if scanned != 2 {
+		t.Errorf("expected 2 scanned, got %d", scanned)
+	}
+	if len(expired) != 1 {
+		t.Fatalf("expected 1 expired, got %d", len(expired))
+	}
+	if expired[0].name != "old-file.txt" {
+		t.Errorf("expected old-file.txt expired, got %s", expired[0].name)
+	}
+}
+
+func TestIntegration_ListExpiredObjectsByRules_TagFilter(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "tag-bucket"
+	bucketDir := bucketsPath + "/" + bucket
+
+	old := time.Now().Add(-60 * 24 * time.Hour)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+
+	// Object with matching tag.
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name:       "tagged.txt",
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 100},
+		Extended:   map[string][]byte{"X-Amz-Tagging-env": []byte("dev")},
+	})
+	// Object without tag.
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name:       "untagged.txt",
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 100},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "tag-expire", Status: "Enabled",
+		ExpirationDays: 30,
+		FilterTags:     map[string]string{"env": "dev"},
+	}}
+
+	expired, _, err := listExpiredObjectsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("listExpiredObjectsByRules: %v", err)
+	}
+
+	if len(expired) != 1 {
+		t.Fatalf("expected 1 expired (tagged only), got %d", len(expired))
+	}
+	if expired[0].name != "tagged.txt" {
+		t.Errorf("expected tagged.txt, got %s", expired[0].name)
+	}
+}
+
+func TestIntegration_ProcessVersionsDirectory(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "versioned-bucket"
+	bucketDir := bucketsPath + "/" + bucket
+	versionsDir := bucketDir + "/key.versions"
+
+	now := time.Now()
+	t1 := now.Add(-90 * 24 * time.Hour) // oldest
+	t2 := now.Add(-60 * 24 * time.Hour)
+	t3 := now.Add(-1 * 24 * time.Hour)  // newest (latest)
+
+	vid1 := testVersionId(t1)
+	vid2 := testVersionId(t2)
+	vid3 := testVersionId(t3)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "key.versions", IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey: []byte(vid3),
+		},
+	})
+
+	// Three versions: vid3 (latest), vid2 (noncurrent), vid1 (noncurrent)
+	server.putEntry(versionsDir, &filer_pb.Entry{
+		Name:       "v_" + vid3,
+		Attributes: &filer_pb.FuseAttributes{Mtime: t3.Unix(), FileSize: 100},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(vid3),
+		},
+	})
+	server.putEntry(versionsDir, &filer_pb.Entry{
+		Name:       "v_" + vid2,
+		Attributes: &filer_pb.FuseAttributes{Mtime: t2.Unix(), FileSize: 100},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(vid2),
+		},
+	})
+	server.putEntry(versionsDir, &filer_pb.Entry{
+		Name:       "v_" + vid1,
+		Attributes: &filer_pb.FuseAttributes{Mtime: t1.Unix(), FileSize: 100},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(vid1),
+		},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "noncurrent-30d", Status: "Enabled",
+		NoncurrentVersionExpirationDays: 30,
+	}}
+
+	expired, scanned, err := processVersionsDirectory(
+		context.Background(), client, versionsDir, bucketDir,
+		rules, now, false, 100,
+	)
+	if err != nil {
+		t.Fatalf("processVersionsDirectory: %v", err)
+	}
+
+	// vid3 is latest (not expired). vid2 became noncurrent when vid3 was created
+	// (1 day ago), so vid2 is NOT old enough (< 30 days noncurrent).
+	// vid1 became noncurrent when vid2 was created (60 days ago), so vid1 IS expired.
+	if scanned != 2 {
+		t.Errorf("expected 2 scanned (non-current versions), got %d", scanned)
+	}
+	if len(expired) != 1 {
+		t.Fatalf("expected 1 expired (only vid1), got %d", len(expired))
+	}
+	if expired[0].name != "v_"+vid1 {
+		t.Errorf("expected v_%s expired, got %s", vid1, expired[0].name)
+	}
+}
+
+func TestIntegration_ProcessVersionsDirectory_NewerNoncurrentVersions(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "keep-n-bucket"
+	bucketDir := bucketsPath + "/" + bucket
+	versionsDir := bucketDir + "/obj.versions"
+
+	now := time.Now()
+	// Create 5 versions, all old enough to expire by days alone.
+	versions := make([]time.Time, 5)
+	vids := make([]string, 5)
+	for i := 0; i < 5; i++ {
+		versions[i] = now.Add(time.Duration(-(90 - i*10)) * 24 * time.Hour)
+		vids[i] = testVersionId(versions[i])
+	}
+	// vids[4] is newest (latest), vids[0] is oldest
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "obj.versions", IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey: []byte(vids[4]),
+		},
+	})
+
+	for i, vid := range vids {
+		server.putEntry(versionsDir, &filer_pb.Entry{
+			Name:       "v_" + vid,
+			Attributes: &filer_pb.FuseAttributes{Mtime: versions[i].Unix(), FileSize: 100},
+			Extended: map[string][]byte{
+				s3_constants.ExtVersionIdKey: []byte(vid),
+			},
+		})
+	}
+
+	rules := []s3lifecycle.Rule{{
+		ID: "keep-2", Status: "Enabled",
+		NoncurrentVersionExpirationDays: 7,
+		NewerNoncurrentVersions:         2,
+	}}
+
+	expired, _, err := processVersionsDirectory(
+		context.Background(), client, versionsDir, bucketDir,
+		rules, now, false, 100,
+	)
+	if err != nil {
+		t.Fatalf("processVersionsDirectory: %v", err)
+	}
+
+	// 4 noncurrent versions (vids[0..3]). Keep newest 2 (vids[3], vids[2]).
+	// Expire vids[1] and vids[0].
+	if len(expired) != 2 {
+		t.Fatalf("expected 2 expired (keep 2 newest noncurrent), got %d", len(expired))
+	}
+	expiredNames := map[string]bool{}
+	for _, e := range expired {
+		expiredNames[e.name] = true
+	}
+	if !expiredNames["v_"+vids[0]] {
+		t.Errorf("expected vids[0] (oldest) to be expired")
+	}
+	if !expiredNames["v_"+vids[1]] {
+		t.Errorf("expected vids[1] to be expired")
+	}
+}
+
+func TestIntegration_AbortMPUsByRules(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "mpu-bucket"
+	uploadsDir := bucketsPath + "/" + bucket + "/.uploads"
+
+	now := time.Now()
+	old := now.Add(-10 * 24 * time.Hour)
+	recent := now.Add(-2 * 24 * time.Hour)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+	server.putEntry(bucketsPath+"/"+bucket, &filer_pb.Entry{Name: ".uploads", IsDirectory: true})
+
+	// Old upload under logs/ prefix.
+	server.putEntry(uploadsDir, &filer_pb.Entry{
+		Name: "logs_upload1", IsDirectory: true,
+		Attributes: &filer_pb.FuseAttributes{Crtime: old.Unix()},
+	})
+	// Recent upload under logs/ prefix.
+	server.putEntry(uploadsDir, &filer_pb.Entry{
+		Name: "logs_upload2", IsDirectory: true,
+		Attributes: &filer_pb.FuseAttributes{Crtime: recent.Unix()},
+	})
+	// Old upload under data/ prefix (should not match logs/ rule).
+	server.putEntry(uploadsDir, &filer_pb.Entry{
+		Name: "data_upload1", IsDirectory: true,
+		Attributes: &filer_pb.FuseAttributes{Crtime: old.Unix()},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "abort-logs", Status: "Enabled",
+		Prefix:                      "logs",
+		AbortMPUDaysAfterInitiation: 7,
+	}}
+
+	aborted, errs, err := abortMPUsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("abortMPUsByRules: %v", err)
+	}
+	if errs != 0 {
+		t.Errorf("expected 0 errors, got %d", errs)
+	}
+
+	// Only logs_upload1 should be aborted (old + matches prefix).
+	// logs_upload2 is too recent, data_upload1 doesn't match prefix.
+	if aborted != 1 {
+		t.Errorf("expected 1 aborted, got %d", aborted)
+	}
+
+	// Verify the correct upload was removed.
+	if server.hasEntry(uploadsDir, "logs_upload1") {
+		t.Error("logs_upload1 should have been removed")
+	}
+	if !server.hasEntry(uploadsDir, "logs_upload2") {
+		t.Error("logs_upload2 should still exist")
+	}
+	if !server.hasEntry(uploadsDir, "data_upload1") {
+		t.Error("data_upload1 should still exist (wrong prefix)")
+	}
+}
+
+func TestIntegration_DeleteExpiredObjects(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "delete-bucket"
+	bucketDir := bucketsPath + "/" + bucket
+
+	now := time.Now()
+	old := now.Add(-60 * 24 * time.Hour)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name:       "to-delete.txt",
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 100},
+	})
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name:       "to-keep.txt",
+		Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileSize: 100},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "expire", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+
+	expired, _, err := listExpiredObjectsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+
+	// Actually delete them.
+	deleted, errs, err := deleteExpiredObjects(context.Background(), client, expired)
+	if err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	if deleted != 1 || errs != 0 {
+		t.Errorf("expected 1 deleted 0 errors, got %d deleted %d errors", deleted, errs)
+	}
+
+	if server.hasEntry(bucketDir, "to-delete.txt") {
+		t.Error("to-delete.txt should have been removed")
+	}
+	if !server.hasEntry(bucketDir, "to-keep.txt") {
+		t.Error("to-keep.txt should still exist")
+	}
+}
+
+// TestIntegration_VersionedBucket_ExpirationDays verifies that Expiration.Days
+// rules correctly detect and delete the latest version in a versioned bucket
+// where all data lives in .versions/ directories (issue #8757).
+func TestIntegration_VersionedBucket_ExpirationDays(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "versioned-expire"
+	bucketDir := bucketsPath + "/" + bucket
+
+	now := time.Now()
+	old := now.Add(-60 * 24 * time.Hour)    // 60 days ago — should expire
+	recent := now.Add(-5 * 24 * time.Hour)  // 5 days ago — should NOT expire
+
+	vidOld := testVersionId(old)
+	vidRecent := testVersionId(recent)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+
+	// --- Single-version object (old, should expire) ---
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "old-file.txt" + s3_constants.VersionsFolder, IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey:         []byte(vidOld),
+			s3_constants.ExtLatestVersionFileNameKey:    []byte("v_" + vidOld),
+			s3_constants.ExtLatestVersionMtimeKey:       []byte(strconv.FormatInt(old.Unix(), 10)),
+			s3_constants.ExtLatestVersionSizeKey:        []byte("3400000000"),
+			s3_constants.ExtLatestVersionIsDeleteMarker: []byte("false"),
+		},
+	})
+	oldVersionsDir := bucketDir + "/old-file.txt" + s3_constants.VersionsFolder
+	server.putEntry(oldVersionsDir, &filer_pb.Entry{
+		Name:       "v_" + vidOld,
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 3400000000},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(vidOld),
+		},
+	})
+
+	// --- Single-version object (recent, should NOT expire) ---
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "recent-file.txt" + s3_constants.VersionsFolder, IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey:         []byte(vidRecent),
+			s3_constants.ExtLatestVersionFileNameKey:    []byte("v_" + vidRecent),
+			s3_constants.ExtLatestVersionMtimeKey:       []byte(strconv.FormatInt(recent.Unix(), 10)),
+			s3_constants.ExtLatestVersionSizeKey:        []byte("3400000000"),
+			s3_constants.ExtLatestVersionIsDeleteMarker: []byte("false"),
+		},
+	})
+	recentVersionsDir := bucketDir + "/recent-file.txt" + s3_constants.VersionsFolder
+	server.putEntry(recentVersionsDir, &filer_pb.Entry{
+		Name:       "v_" + vidRecent,
+		Attributes: &filer_pb.FuseAttributes{Mtime: recent.Unix(), FileSize: 3400000000},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(vidRecent),
+		},
+	})
+
+	// --- Object with delete marker as latest (should NOT be expired by Expiration.Days) ---
+	vidMarker := testVersionId(old)
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "deleted-obj.txt" + s3_constants.VersionsFolder, IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey:         []byte(vidMarker),
+			s3_constants.ExtLatestVersionFileNameKey:    []byte("v_" + vidMarker),
+			s3_constants.ExtLatestVersionMtimeKey:       []byte(strconv.FormatInt(old.Unix(), 10)),
+			s3_constants.ExtLatestVersionIsDeleteMarker: []byte("true"),
+		},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "expire-30d", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+
+	expired, scanned, err := listExpiredObjectsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("listExpiredObjectsByRules: %v", err)
+	}
+
+	// Only old-file.txt's latest version should be expired.
+	// recent-file.txt is too young; deleted-obj.txt is a delete marker.
+	if len(expired) != 1 {
+		t.Fatalf("expected 1 expired, got %d: %+v", len(expired), expired)
+	}
+	if expired[0].dir != oldVersionsDir {
+		t.Errorf("expected dir=%s, got %s", oldVersionsDir, expired[0].dir)
+	}
+	if expired[0].name != "v_"+vidOld {
+		t.Errorf("expected name=v_%s, got %s", vidOld, expired[0].name)
+	}
+	// The old-file.txt latest version should count as scanned.
+	if scanned < 1 {
+		t.Errorf("expected at least 1 scanned, got %d", scanned)
+	}
+}
+
+// TestIntegration_VersionedBucket_ExpirationDays_DeleteAndCleanup verifies
+// end-to-end deletion and .versions directory cleanup for a single-version
+// versioned object expired by Expiration.Days.
+func TestIntegration_VersionedBucket_ExpirationDays_DeleteAndCleanup(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "versioned-cleanup"
+	bucketDir := bucketsPath + "/" + bucket
+
+	now := time.Now()
+	old := now.Add(-60 * 24 * time.Hour)
+	vidOld := testVersionId(old)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+
+	// Single-version object that should expire.
+	versionsDir := bucketDir + "/data.bin" + s3_constants.VersionsFolder
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "data.bin" + s3_constants.VersionsFolder, IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey:         []byte(vidOld),
+			s3_constants.ExtLatestVersionFileNameKey:    []byte("v_" + vidOld),
+			s3_constants.ExtLatestVersionMtimeKey:       []byte(strconv.FormatInt(old.Unix(), 10)),
+			s3_constants.ExtLatestVersionSizeKey:        []byte("1024"),
+			s3_constants.ExtLatestVersionIsDeleteMarker: []byte("false"),
+		},
+	})
+	server.putEntry(versionsDir, &filer_pb.Entry{
+		Name:       "v_" + vidOld,
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 1024},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(vidOld),
+		},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "expire-30d", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+
+	// Step 1: Detect expired.
+	expired, _, err := listExpiredObjectsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(expired) != 1 {
+		t.Fatalf("expected 1 expired, got %d", len(expired))
+	}
+
+	// Step 2: Delete the expired version file.
+	deleted, errs, delErr := deleteExpiredObjects(context.Background(), client, expired)
+	if delErr != nil {
+		t.Fatalf("delete: %v", delErr)
+	}
+	if deleted != 1 || errs != 0 {
+		t.Errorf("expected 1 deleted 0 errors, got %d deleted %d errors", deleted, errs)
+	}
+
+	// Version file should be gone.
+	if server.hasEntry(versionsDir, "v_"+vidOld) {
+		t.Error("version file should have been removed")
+	}
+
+	// Step 3: Cleanup empty .versions directory.
+	cleaned := cleanupEmptyVersionsDirectories(context.Background(), client, expired)
+	if cleaned != 1 {
+		t.Errorf("expected 1 directory cleaned, got %d", cleaned)
+	}
+
+	// The .versions directory itself should be gone.
+	if server.hasEntry(bucketDir, "data.bin"+s3_constants.VersionsFolder) {
+		t.Error(".versions directory should have been removed after cleanup")
+	}
+}
+
+// TestIntegration_VersionedBucket_MultiVersion_ExpirationDays verifies that
+// when a multi-version object's latest version expires, only the latest
+// version is deleted and noncurrent versions remain.
+func TestIntegration_VersionedBucket_MultiVersion_ExpirationDays(t *testing.T) {
+	server, client := startTestFiler(t)
+	bucketsPath := "/buckets"
+	bucket := "versioned-multi"
+	bucketDir := bucketsPath + "/" + bucket
+
+	now := time.Now()
+	tOld := now.Add(-60 * 24 * time.Hour)
+	tNoncurrent := now.Add(-90 * 24 * time.Hour)
+	vidLatest := testVersionId(tOld)
+	vidNoncurrent := testVersionId(tNoncurrent)
+
+	server.putEntry(bucketsPath, &filer_pb.Entry{Name: bucket, IsDirectory: true})
+
+	versionsDir := bucketDir + "/multi.txt" + s3_constants.VersionsFolder
+	server.putEntry(bucketDir, &filer_pb.Entry{
+		Name: "multi.txt" + s3_constants.VersionsFolder, IsDirectory: true,
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionIdKey:         []byte(vidLatest),
+			s3_constants.ExtLatestVersionFileNameKey:    []byte("v_" + vidLatest),
+			s3_constants.ExtLatestVersionMtimeKey:       []byte(strconv.FormatInt(tOld.Unix(), 10)),
+			s3_constants.ExtLatestVersionSizeKey:        []byte("500"),
+			s3_constants.ExtLatestVersionIsDeleteMarker: []byte("false"),
+		},
+	})
+	server.putEntry(versionsDir, &filer_pb.Entry{
+		Name:       "v_" + vidLatest,
+		Attributes: &filer_pb.FuseAttributes{Mtime: tOld.Unix(), FileSize: 500},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte(vidLatest)},
+	})
+	server.putEntry(versionsDir, &filer_pb.Entry{
+		Name:       "v_" + vidNoncurrent,
+		Attributes: &filer_pb.FuseAttributes{Mtime: tNoncurrent.Unix(), FileSize: 500},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte(vidNoncurrent)},
+	})
+
+	rules := []s3lifecycle.Rule{{
+		ID: "expire-30d", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+
+	expired, _, err := listExpiredObjectsByRules(context.Background(), client, bucketsPath, bucket, rules, 100)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	// Only the latest version should be detected as expired.
+	if len(expired) != 1 {
+		t.Fatalf("expected 1 expired (latest only), got %d", len(expired))
+	}
+	if expired[0].name != "v_"+vidLatest {
+		t.Errorf("expected latest version expired, got %s", expired[0].name)
+	}
+
+	// Delete it.
+	deleted, errs, delErr := deleteExpiredObjects(context.Background(), client, expired)
+	if delErr != nil {
+		t.Fatalf("delete: %v", delErr)
+	}
+	if deleted != 1 || errs != 0 {
+		t.Errorf("expected 1 deleted 0 errors, got %d deleted %d errors", deleted, errs)
+	}
+
+	// Noncurrent version should still exist.
+	if !server.hasEntry(versionsDir, "v_"+vidNoncurrent) {
+		t.Error("noncurrent version should still exist")
+	}
+
+	// .versions directory should NOT be cleaned up (not empty).
+	cleaned := cleanupEmptyVersionsDirectories(context.Background(), client, expired)
+	if cleaned != 0 {
+		t.Errorf("expected 0 directories cleaned (not empty), got %d", cleaned)
+	}
+	if !server.hasEntry(bucketDir, "multi.txt"+s3_constants.VersionsFolder) {
+		t.Error(".versions directory should still exist (has noncurrent version)")
+	}
+}

weed/plugin/worker/lifecycle/rules.go

@@ -0,0 +1,199 @@
+package lifecycle
+
+import (
+	"bytes"
+	"context"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+)
+
+// lifecycleConfig mirrors the XML structure just enough to parse rules.
+// We define a minimal local struct to avoid importing the s3api package
+// (which would create a circular dependency if s3api ever imports the worker).
+type lifecycleConfig struct {
+	XMLName xml.Name              `xml:"LifecycleConfiguration"`
+	Rules   []lifecycleConfigRule `xml:"Rule"`
+}
+
+type lifecycleConfigRule struct {
+	ID         string                        `xml:"ID"`
+	Status     string                        `xml:"Status"`
+	Filter     lifecycleFilter               `xml:"Filter"`
+	Prefix     string                        `xml:"Prefix"`
+	Expiration lifecycleExpiration            `xml:"Expiration"`
+	NoncurrentVersionExpiration noncurrentVersionExpiration `xml:"NoncurrentVersionExpiration"`
+	AbortIncompleteMultipartUpload abortMPU   `xml:"AbortIncompleteMultipartUpload"`
+}
+
+type lifecycleFilter struct {
+	Prefix                string          `xml:"Prefix"`
+	Tag                   lifecycleTag    `xml:"Tag"`
+	And                   lifecycleAnd    `xml:"And"`
+	ObjectSizeGreaterThan int64           `xml:"ObjectSizeGreaterThan"`
+	ObjectSizeLessThan    int64           `xml:"ObjectSizeLessThan"`
+}
+
+type lifecycleAnd struct {
+	Prefix                string          `xml:"Prefix"`
+	Tags                  []lifecycleTag  `xml:"Tag"`
+	ObjectSizeGreaterThan int64           `xml:"ObjectSizeGreaterThan"`
+	ObjectSizeLessThan    int64           `xml:"ObjectSizeLessThan"`
+}
+
+type lifecycleTag struct {
+	Key   string `xml:"Key"`
+	Value string `xml:"Value"`
+}
+
+type lifecycleExpiration struct {
+	Days                       int    `xml:"Days"`
+	Date                       string `xml:"Date"`
+	ExpiredObjectDeleteMarker  bool   `xml:"ExpiredObjectDeleteMarker"`
+}
+
+type noncurrentVersionExpiration struct {
+	NoncurrentDays          int `xml:"NoncurrentDays"`
+	NewerNoncurrentVersions int `xml:"NewerNoncurrentVersions"`
+}
+
+type abortMPU struct {
+	DaysAfterInitiation int `xml:"DaysAfterInitiation"`
+}
+
+// errMalformedLifecycleXML indicates the lifecycle XML exists but could not be parsed.
+// Callers should fail closed (not fall back to TTL) to avoid broader deletions.
+var errMalformedLifecycleXML = errors.New("malformed lifecycle XML")
+
+// loadLifecycleRulesFromBucket reads the lifecycle XML from a bucket's
+// metadata and converts it to evaluator-friendly rules.
+//
+// Returns:
+//   - (rules, nil) when lifecycle XML is configured and parseable
+//   - (nil, nil) when no lifecycle XML is configured (caller should use TTL fallback)
+//   - (nil, errMalformedLifecycleXML) when XML exists but is malformed (fail closed)
+//   - (nil, err) for transient filer errors (caller should use TTL fallback with warning)
+func loadLifecycleRulesFromBucket(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	bucketsPath, bucket string,
+) ([]s3lifecycle.Rule, error) {
+	bucketDir := bucketsPath
+	resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: bucketDir,
+		Name:      bucket,
+	})
+	if err != nil {
+		// Transient filer error — not the same as malformed XML.
+		return nil, fmt.Errorf("lookup bucket %s: %w", bucket, err)
+	}
+	if resp.Entry == nil || resp.Entry.Extended == nil {
+		return nil, nil
+	}
+	xmlData := resp.Entry.Extended[lifecycleXMLKey]
+	if len(xmlData) == 0 {
+		return nil, nil
+	}
+	rules, parseErr := parseLifecycleXML(xmlData)
+	if parseErr != nil {
+		return nil, fmt.Errorf("%w: bucket %s: %v", errMalformedLifecycleXML, bucket, parseErr)
+	}
+	// Return non-nil empty slice when XML was present but yielded no rules
+	// (e.g., all rules disabled). This lets callers distinguish "no XML" (nil)
+	// from "XML present, no effective rules" (empty slice).
+	if rules == nil {
+		rules = []s3lifecycle.Rule{}
+	}
+	return rules, nil
+}
+
+// parseLifecycleXML parses lifecycle configuration XML and converts it
+// to evaluator-friendly rules.
+func parseLifecycleXML(data []byte) ([]s3lifecycle.Rule, error) {
+	var config lifecycleConfig
+	if err := xml.NewDecoder(bytes.NewReader(data)).Decode(&config); err != nil {
+		return nil, fmt.Errorf("decode lifecycle XML: %w", err)
+	}
+
+	var rules []s3lifecycle.Rule
+	for _, r := range config.Rules {
+		rule := s3lifecycle.Rule{
+			ID:     r.ID,
+			Status: r.Status,
+		}
+
+		// Resolve prefix: Filter.And.Prefix > Filter.Prefix > Rule.Prefix
+		switch {
+		case r.Filter.And.Prefix != "" || len(r.Filter.And.Tags) > 0 ||
+			r.Filter.And.ObjectSizeGreaterThan > 0 || r.Filter.And.ObjectSizeLessThan > 0:
+			rule.Prefix = r.Filter.And.Prefix
+			rule.FilterTags = tagsToMap(r.Filter.And.Tags)
+			rule.FilterSizeGreaterThan = r.Filter.And.ObjectSizeGreaterThan
+			rule.FilterSizeLessThan = r.Filter.And.ObjectSizeLessThan
+		case r.Filter.Tag.Key != "":
+			rule.Prefix = r.Filter.Prefix
+			rule.FilterTags = map[string]string{r.Filter.Tag.Key: r.Filter.Tag.Value}
+			rule.FilterSizeGreaterThan = r.Filter.ObjectSizeGreaterThan
+			rule.FilterSizeLessThan = r.Filter.ObjectSizeLessThan
+		default:
+			if r.Filter.Prefix != "" {
+				rule.Prefix = r.Filter.Prefix
+			} else {
+				rule.Prefix = r.Prefix
+			}
+			rule.FilterSizeGreaterThan = r.Filter.ObjectSizeGreaterThan
+			rule.FilterSizeLessThan = r.Filter.ObjectSizeLessThan
+		}
+
+		rule.ExpirationDays = r.Expiration.Days
+		rule.ExpiredObjectDeleteMarker = r.Expiration.ExpiredObjectDeleteMarker
+		rule.NoncurrentVersionExpirationDays = r.NoncurrentVersionExpiration.NoncurrentDays
+		rule.NewerNoncurrentVersions = r.NoncurrentVersionExpiration.NewerNoncurrentVersions
+		rule.AbortMPUDaysAfterInitiation = r.AbortIncompleteMultipartUpload.DaysAfterInitiation
+
+		// Parse Date if present.
+		if r.Expiration.Date != "" {
+			// Date may be RFC3339 or ISO 8601 date-only.
+			parsed, parseErr := parseExpirationDate(r.Expiration.Date)
+			if parseErr != nil {
+				glog.V(1).Infof("s3_lifecycle: skipping rule %s: invalid expiration date %q: %v", r.ID, r.Expiration.Date, parseErr)
+				continue
+			}
+			rule.ExpirationDate = parsed
+		}
+
+		rules = append(rules, rule)
+	}
+	return rules, nil
+}
+
+func tagsToMap(tags []lifecycleTag) map[string]string {
+	if len(tags) == 0 {
+		return nil
+	}
+	m := make(map[string]string, len(tags))
+	for _, t := range tags {
+		m[t.Key] = t.Value
+	}
+	return m
+}
+
+func parseExpirationDate(s string) (time.Time, error) {
+	// Try RFC3339 first, then ISO 8601 date-only.
+	formats := []string{
+		"2006-01-02T15:04:05Z07:00",
+		"2006-01-02",
+	}
+	for _, f := range formats {
+		t, err := time.Parse(f, s)
+		if err == nil {
+			return t, nil
+		}
+	}
+	return time.Time{}, fmt.Errorf("unrecognized date format: %s", s)
+}

weed/plugin/worker/lifecycle/rules_test.go

@@ -0,0 +1,256 @@
+package lifecycle
+
+import (
+	"testing"
+	"time"
+)
+
+func TestParseLifecycleXML_CompleteConfig(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>rotation</ID>
+    <Filter><Prefix></Prefix></Filter>
+    <Status>Enabled</Status>
+    <Expiration><Days>30</Days></Expiration>
+    <NoncurrentVersionExpiration>
+      <NoncurrentDays>7</NoncurrentDays>
+      <NewerNoncurrentVersions>2</NewerNoncurrentVersions>
+    </NoncurrentVersionExpiration>
+    <AbortIncompleteMultipartUpload>
+      <DaysAfterInitiation>3</DaysAfterInitiation>
+    </AbortIncompleteMultipartUpload>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if len(rules) != 1 {
+		t.Fatalf("expected 1 rule, got %d", len(rules))
+	}
+
+	r := rules[0]
+	if r.ID != "rotation" {
+		t.Errorf("expected ID 'rotation', got %q", r.ID)
+	}
+	if r.Status != "Enabled" {
+		t.Errorf("expected Status 'Enabled', got %q", r.Status)
+	}
+	if r.ExpirationDays != 30 {
+		t.Errorf("expected ExpirationDays=30, got %d", r.ExpirationDays)
+	}
+	if r.NoncurrentVersionExpirationDays != 7 {
+		t.Errorf("expected NoncurrentVersionExpirationDays=7, got %d", r.NoncurrentVersionExpirationDays)
+	}
+	if r.NewerNoncurrentVersions != 2 {
+		t.Errorf("expected NewerNoncurrentVersions=2, got %d", r.NewerNoncurrentVersions)
+	}
+	if r.AbortMPUDaysAfterInitiation != 3 {
+		t.Errorf("expected AbortMPUDaysAfterInitiation=3, got %d", r.AbortMPUDaysAfterInitiation)
+	}
+}
+
+func TestParseLifecycleXML_PrefixFilter(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>logs</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix>logs/</Prefix></Filter>
+    <Expiration><Days>7</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if len(rules) != 1 {
+		t.Fatalf("expected 1 rule, got %d", len(rules))
+	}
+	if rules[0].Prefix != "logs/" {
+		t.Errorf("expected Prefix='logs/', got %q", rules[0].Prefix)
+	}
+}
+
+func TestParseLifecycleXML_LegacyPrefix(t *testing.T) {
+	// Old-style <Prefix> at rule level instead of inside <Filter>.
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>old</ID>
+    <Status>Enabled</Status>
+    <Prefix>archive/</Prefix>
+    <Expiration><Days>90</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if len(rules) != 1 {
+		t.Fatalf("expected 1 rule, got %d", len(rules))
+	}
+	if rules[0].Prefix != "archive/" {
+		t.Errorf("expected Prefix='archive/', got %q", rules[0].Prefix)
+	}
+}
+
+func TestParseLifecycleXML_TagFilter(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>tag-rule</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <Tag><Key>env</Key><Value>dev</Value></Tag>
+    </Filter>
+    <Expiration><Days>1</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if len(rules) != 1 {
+		t.Fatalf("expected 1 rule, got %d", len(rules))
+	}
+	r := rules[0]
+	if len(r.FilterTags) != 1 || r.FilterTags["env"] != "dev" {
+		t.Errorf("expected FilterTags={env:dev}, got %v", r.FilterTags)
+	}
+}
+
+func TestParseLifecycleXML_AndFilter(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>and-rule</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <And>
+        <Prefix>data/</Prefix>
+        <Tag><Key>env</Key><Value>staging</Value></Tag>
+        <ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan>
+      </And>
+    </Filter>
+    <Expiration><Days>14</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if len(rules) != 1 {
+		t.Fatalf("expected 1 rule, got %d", len(rules))
+	}
+	r := rules[0]
+	if r.Prefix != "data/" {
+		t.Errorf("expected Prefix='data/', got %q", r.Prefix)
+	}
+	if r.FilterTags["env"] != "staging" {
+		t.Errorf("expected tag env=staging, got %v", r.FilterTags)
+	}
+	if r.FilterSizeGreaterThan != 1024 {
+		t.Errorf("expected FilterSizeGreaterThan=1024, got %d", r.FilterSizeGreaterThan)
+	}
+}
+
+func TestParseLifecycleXML_ExpirationDate(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>date-rule</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <Expiration><Date>2026-06-01T00:00:00Z</Date></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	expected := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)
+	if !rules[0].ExpirationDate.Equal(expected) {
+		t.Errorf("expected ExpirationDate=%v, got %v", expected, rules[0].ExpirationDate)
+	}
+}
+
+func TestParseLifecycleXML_ExpiredObjectDeleteMarker(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>marker-cleanup</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if !rules[0].ExpiredObjectDeleteMarker {
+		t.Error("expected ExpiredObjectDeleteMarker=true")
+	}
+}
+
+func TestParseLifecycleXML_MultipleRules(t *testing.T) {
+	xml := []byte(`<LifecycleConfiguration>
+  <Rule>
+    <ID>rule1</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix>logs/</Prefix></Filter>
+    <Expiration><Days>7</Days></Expiration>
+  </Rule>
+  <Rule>
+    <ID>rule2</ID>
+    <Status>Disabled</Status>
+    <Filter><Prefix>temp/</Prefix></Filter>
+    <Expiration><Days>1</Days></Expiration>
+  </Rule>
+  <Rule>
+    <ID>rule3</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <Expiration><Days>365</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`)
+
+	rules, err := parseLifecycleXML(xml)
+	if err != nil {
+		t.Fatalf("parseLifecycleXML: %v", err)
+	}
+	if len(rules) != 3 {
+		t.Fatalf("expected 3 rules, got %d", len(rules))
+	}
+	if rules[1].Status != "Disabled" {
+		t.Errorf("expected rule2 Status=Disabled, got %q", rules[1].Status)
+	}
+}
+
+func TestParseExpirationDate(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   string
+		want    time.Time
+		wantErr bool
+	}{
+		{"rfc3339_utc", "2026-06-01T00:00:00Z", time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC), false},
+		{"rfc3339_offset", "2026-06-01T00:00:00+05:00", time.Date(2026, 6, 1, 0, 0, 0, 0, time.FixedZone("", 5*3600)), false},
+		{"date_only", "2026-06-01", time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC), false},
+		{"invalid", "not-a-date", time.Time{}, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := parseExpirationDate(tt.input)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("parseExpirationDate(%q) error = %v, wantErr %v", tt.input, err, tt.wantErr)
+				return
+			}
+			if !tt.wantErr && !got.Equal(tt.want) {
+				t.Errorf("parseExpirationDate(%q) = %v, want %v", tt.input, got, tt.want)
+			}
+		})
+	}
+}

weed/plugin/worker/lifecycle/version_test.go

@@ -0,0 +1,112 @@
+package lifecycle
+
+import (
+	"fmt"
+	"math"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+)
+
+// makeVersionId creates a new-format version ID from a timestamp.
+func makeVersionId(t time.Time) string {
+	inverted := math.MaxInt64 - t.UnixNano()
+	return fmt.Sprintf("%016x", inverted) + "0000000000000000"
+}
+
+func TestSortVersionsByVersionId(t *testing.T) {
+	t1 := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	t2 := time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC)
+	t3 := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
+
+	vid1 := makeVersionId(t1)
+	vid2 := makeVersionId(t2)
+	vid3 := makeVersionId(t3)
+
+	entries := []*filer_pb.Entry{
+		{Name: "v_" + vid1},
+		{Name: "v_" + vid3},
+		{Name: "v_" + vid2},
+	}
+
+	sortVersionsByVersionId(entries)
+
+	// Should be sorted newest first: t3, t2, t1.
+	expected := []string{"v_" + vid3, "v_" + vid2, "v_" + vid1}
+	for i, want := range expected {
+		if entries[i].Name != want {
+			t.Errorf("entries[%d].Name = %s, want %s", i, entries[i].Name, want)
+		}
+	}
+}
+
+func TestSortVersionsByVersionId_SameTimestampDifferentSuffix(t *testing.T) {
+	// Two versions with the same timestamp prefix but different random suffix.
+	// The sort must still produce a deterministic order.
+	base := makeVersionId(time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC))
+	vid1 := base[:16] + "aaaaaaaaaaaaaaaa"
+	vid2 := base[:16] + "bbbbbbbbbbbbbbbb"
+
+	entries := []*filer_pb.Entry{
+		{Name: "v_" + vid2},
+		{Name: "v_" + vid1},
+	}
+
+	sortVersionsByVersionId(entries)
+
+	// New format: smaller hex = newer. vid1 ("aaa...") < vid2 ("bbb...") so vid1 is newer.
+	if strings.TrimPrefix(entries[0].Name, "v_") != vid1 {
+		t.Errorf("expected vid1 (newer) first, got %s", entries[0].Name)
+	}
+}
+
+func TestCompareVersionIdsMixedFormats(t *testing.T) {
+	// Old format: raw nanosecond timestamp (below threshold ~0x17...).
+	// New format: inverted timestamp (above threshold ~0x68...).
+	oldTs := time.Date(2023, 6, 15, 12, 0, 0, 0, time.UTC)
+	newTs := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
+
+	oldFormatId := fmt.Sprintf("%016x", oldTs.UnixNano()) + "abcdef0123456789"
+	newFormatId := makeVersionId(newTs) // uses inverted timestamp
+
+	// newTs is more recent, so newFormatId should sort as "newer".
+	cmp := s3lifecycle.CompareVersionIds(newFormatId, oldFormatId)
+	if cmp >= 0 {
+		t.Errorf("expected new-format ID (2026) to be newer than old-format ID (2023), got cmp=%d", cmp)
+	}
+
+	// Reverse comparison.
+	cmp2 := s3lifecycle.CompareVersionIds(oldFormatId, newFormatId)
+	if cmp2 <= 0 {
+		t.Errorf("expected old-format ID (2023) to be older than new-format ID (2026), got cmp=%d", cmp2)
+	}
+
+	// Sort a mixed slice: should be newest-first.
+	entries := []*filer_pb.Entry{
+		{Name: "v_" + oldFormatId},
+		{Name: "v_" + newFormatId},
+	}
+	sortVersionsByVersionId(entries)
+
+	if strings.TrimPrefix(entries[0].Name, "v_") != newFormatId {
+		t.Errorf("expected new-format (newer) entry first after sort")
+	}
+}
+
+func TestVersionsDirectoryNaming(t *testing.T) {
+	if s3_constants.VersionsFolder != ".versions" {
+		t.Fatalf("unexpected VersionsFolder constant: %q", s3_constants.VersionsFolder)
+	}
+
+	versionsDir := "/buckets/mybucket/path/to/key.versions"
+	bucketPath := "/buckets/mybucket"
+	relDir := strings.TrimPrefix(versionsDir, bucketPath+"/")
+	objKey := strings.TrimSuffix(relDir, s3_constants.VersionsFolder)
+	if objKey != "path/to/key" {
+		t.Errorf("expected 'path/to/key', got %q", objKey)
+	}
+}

weed/s3api/filer_util.go

@@ -167,12 +167,6 @@ func (s3a *S3ApiServer) exists(parentDirectoryPath string, entryName string, isD
 
 }
 
-func (s3a *S3ApiServer) touch(parentDirectoryPath string, entryName string, entry *filer_pb.Entry) (err error) {
-
-	return filer_pb.Touch(context.Background(), s3a, parentDirectoryPath, entryName, entry)
-
-}
-
 func (s3a *S3ApiServer) getEntry(parentDirectoryPath, entryName string) (entry *filer_pb.Entry, err error) {
 	fullPath := util.NewFullPath(parentDirectoryPath, entryName)
 	return filer_pb.GetEntry(context.Background(), s3a, fullPath)

weed/s3api/s3_bucket_encryption.go

@@ -36,18 +36,6 @@ type ApplyServerSideEncryptionByDefault struct {
 	KMSMasterKeyID string `xml:"KMSMasterKeyID,omitempty"`
 }
 
-// encryptionConfigToProto converts EncryptionConfiguration to protobuf format
-func encryptionConfigToProto(config *s3_pb.EncryptionConfiguration) *s3_pb.EncryptionConfiguration {
-	if config == nil {
-		return nil
-	}
-	return &s3_pb.EncryptionConfiguration{
-		SseAlgorithm:     config.SseAlgorithm,
-		KmsKeyId:         config.KmsKeyId,
-		BucketKeyEnabled: config.BucketKeyEnabled,
-	}
-}
-
 // encryptionConfigFromXML converts XML ServerSideEncryptionConfiguration to protobuf
 func encryptionConfigFromXML(xmlConfig *ServerSideEncryptionConfiguration) *s3_pb.EncryptionConfiguration {
 	if xmlConfig == nil || len(xmlConfig.Rules) == 0 {

weed/s3api/s3_error_utils.go

@@ -7,17 +7,6 @@ import (
 
 // ErrorHandlers provide common error handling patterns for S3 API operations
 
-// handlePutToFilerError logs an error and returns the standard putToFiler error format
-func handlePutToFilerError(operation string, err error, errorCode s3err.ErrorCode) (string, s3err.ErrorCode, string) {
-	glog.Errorf("Failed to %s: %v", operation, err)
-	return "", errorCode, ""
-}
-
-// handlePutToFilerInternalError is a convenience wrapper for internal errors in putToFiler
-func handlePutToFilerInternalError(operation string, err error) (string, s3err.ErrorCode, string) {
-	return handlePutToFilerError(operation, err, s3err.ErrInternalError)
-}
-
 // handleMultipartError logs an error and returns the standard multipart error format
 func handleMultipartError(operation string, err error, errorCode s3err.ErrorCode) (interface{}, s3err.ErrorCode) {
 	glog.Errorf("Failed to %s: %v", operation, err)
@@ -28,27 +17,3 @@ func handleMultipartError(operation string, err error, errorCode s3err.ErrorCode
 func handleMultipartInternalError(operation string, err error) (interface{}, s3err.ErrorCode) {
 	return handleMultipartError(operation, err, s3err.ErrInternalError)
 }
-
-// logErrorAndReturn logs an error with operation context and returns the specified error code
-func logErrorAndReturn(operation string, err error, errorCode s3err.ErrorCode) s3err.ErrorCode {
-	glog.Errorf("Failed to %s: %v", operation, err)
-	return errorCode
-}
-
-// logInternalError is a convenience wrapper for internal error logging
-func logInternalError(operation string, err error) s3err.ErrorCode {
-	return logErrorAndReturn(operation, err, s3err.ErrInternalError)
-}
-
-// SSE-specific error handlers
-
-// handleSSEError handles common SSE-related errors with appropriate context
-func handleSSEError(sseType string, operation string, err error, errorCode s3err.ErrorCode) (string, s3err.ErrorCode, string) {
-	glog.Errorf("Failed to %s for %s: %v", operation, sseType, err)
-	return "", errorCode, ""
-}
-
-// handleSSEInternalError is a convenience wrapper for SSE internal errors
-func handleSSEInternalError(sseType string, operation string, err error) (string, s3err.ErrorCode, string) {
-	return handleSSEError(sseType, operation, err, s3err.ErrInternalError)
-}

weed/s3api/s3_iam_middleware.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -382,84 +381,6 @@ func buildS3ResourceArn(bucket string, objectKey string) string {
 	return "arn:aws:s3:::" + bucket + "/" + objectKey
 }
 
-// hasSpecificQueryParameters checks if the request has query parameters that indicate specific granular operations
-func hasSpecificQueryParameters(query url.Values) bool {
-	// Check for object-level operation indicators
-	objectParams := []string{
-		"acl",        // ACL operations
-		"tagging",    // Tagging operations
-		"retention",  // Object retention
-		"legal-hold", // Legal hold
-		"versions",   // Versioning operations
-	}
-
-	// Check for multipart operation indicators
-	multipartParams := []string{
-		"uploads",    // List/initiate multipart uploads
-		"uploadId",   // Part operations, complete, abort
-		"partNumber", // Upload part
-	}
-
-	// Check for bucket-level operation indicators
-	bucketParams := []string{
-		"policy",         // Bucket policy operations
-		"website",        // Website configuration
-		"cors",           // CORS configuration
-		"lifecycle",      // Lifecycle configuration
-		"notification",   // Event notification
-		"replication",    // Cross-region replication
-		"encryption",     // Server-side encryption
-		"accelerate",     // Transfer acceleration
-		"requestPayment", // Request payment
-		"logging",        // Access logging
-		"versioning",     // Versioning configuration
-		"inventory",      // Inventory configuration
-		"analytics",      // Analytics configuration
-		"metrics",        // CloudWatch metrics
-		"location",       // Bucket location
-	}
-
-	// Check if any of these parameters are present
-	allParams := append(append(objectParams, multipartParams...), bucketParams...)
-	for _, param := range allParams {
-		if _, exists := query[param]; exists {
-			return true
-		}
-	}
-
-	return false
-}
-
-// isMethodActionMismatch detects when HTTP method doesn't align with the intended S3 action
-// This provides a mechanism to use fallback action mapping when there's a semantic mismatch
-func isMethodActionMismatch(method string, fallbackAction Action) bool {
-	switch fallbackAction {
-	case s3_constants.ACTION_WRITE:
-		// WRITE actions should typically use PUT, POST, or DELETE methods
-		// GET/HEAD methods indicate read-oriented operations
-		return method == "GET" || method == "HEAD"
-
-	case s3_constants.ACTION_READ:
-		// READ actions should typically use GET or HEAD methods
-		// PUT, POST, DELETE methods indicate write-oriented operations
-		return method == "PUT" || method == "POST" || method == "DELETE"
-
-	case s3_constants.ACTION_LIST:
-		// LIST actions should typically use GET method
-		// PUT, POST, DELETE methods indicate write-oriented operations
-		return method == "PUT" || method == "POST" || method == "DELETE"
-
-	case s3_constants.ACTION_DELETE_BUCKET:
-		// DELETE_BUCKET should use DELETE method
-		// Other methods indicate different operation types
-		return method != "DELETE"
-
-	default:
-		// For unknown actions or actions that already have s3: prefix, don't assume mismatch
-		return false
-	}
-}
-
 // mapLegacyActionToIAM provides fallback mapping for legacy actions
 // This ensures backward compatibility while the system transitions to granular actions
 func mapLegacyActionToIAM(legacyAction Action) string {
@@ -616,14 +537,6 @@ func ParseUnverifiedJWTToken(tokenString string) (jwt.MapClaims, error) {
 	return nil, fmt.Errorf("invalid token claims")
 }
 
-// minInt returns the minimum of two integers
-func minInt(a, b int) int {
-	if a < b {
-		return a
-	}
-	return b
-}
-
 // SetIAMIntegration adds advanced IAM integration to the S3ApiServer
 func (s3a *S3ApiServer) SetIAMIntegration(iamManager *integration.IAMManager) {
 	if s3a.iam != nil {

weed/s3api/s3_sse_c.go

@@ -272,19 +272,6 @@ func createCTRStreamWithOffset(block cipher.Block, iv []byte, counterOffset uint
 	return stream
 }
 
-// addCounterToIV adds a counter value to the IV (treating last 8 bytes as big-endian counter)
-func addCounterToIV(iv []byte, counter uint64) {
-	// Use the last 8 bytes as a big-endian counter
-	for i := 7; i >= 0; i-- {
-		carry := counter & 0xff
-		iv[len(iv)-8+i] += byte(carry)
-		if iv[len(iv)-8+i] >= byte(carry) {
-			break // No overflow
-		}
-		counter >>= 8
-	}
-}
-
 // GetSourceSSECInfo extracts SSE-C information from source object metadata
 func GetSourceSSECInfo(metadata map[string][]byte) (algorithm string, keyMD5 string, isEncrypted bool) {
 	if alg, exists := metadata[s3_constants.AmzServerSideEncryptionCustomerAlgorithm]; exists {

weed/s3api/s3api_bucket_handlers.go

@@ -952,24 +952,67 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr
 	collectionTtls := fc.GetCollectionTtls(collectionName)
 	changed := false
 
+	// Check whether the bucket has versioning enabled. Versioned buckets must
+	// NOT use the TTL fast-path because:
+	//  1. TTL volumes expire as a unit, destroying all data — including
+	//     noncurrent versions that should be preserved.
+	//  2. Filer-backend TTL (RocksDB compaction, Redis expire) removes entries
+	//     without triggering chunk deletion, leaving orphaned volume data.
+	//  3. On AWS S3, Expiration.Days on a versioned bucket creates a delete
+	//     marker — it does not delete data. TTL has no such nuance.
+	// For versioned buckets the lifecycle worker handles all rule evaluation
+	// at scan time, which correctly operates on individual versions.
+	bucketVersioning, versioningErr := s3a.getBucketVersioningStatus(bucket)
+	if versioningErr != s3err.ErrNone {
+		// Fail closed: if we cannot determine versioning status, treat the
+		// bucket as versioned to avoid creating TTL entries that would
+		// destroy noncurrent versions.
+		glog.V(1).Infof("PutBucketLifecycleConfigurationHandler: could not determine versioning status for %s (err %v), skipping TTL fast-path", bucket, versioningErr)
+	}
+	isVersioned := versioningErr != s3err.ErrNone ||
+		bucketVersioning == s3_constants.VersioningEnabled ||
+		bucketVersioning == s3_constants.VersioningSuspended
+
 	for _, rule := range lifeCycleConfig.Rules {
 		if rule.Status != Enabled {
 			continue
 		}
-		var rulePrefix string
-		switch {
-		case rule.Filter.Prefix.set:
-			rulePrefix = rule.Filter.Prefix.val
-		case rule.Prefix.set:
-			rulePrefix = rule.Prefix.val
-		case !rule.Expiration.Date.IsZero() || rule.Transition.Days > 0 || !rule.Transition.Date.IsZero():
+		// Reject Transition rules — they require storage class migration
+		// infrastructure that does not exist yet.
+		if rule.Transition.set || rule.NoncurrentVersionTransition.set {
 			s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 			return
 		}
 
+		if isVersioned {
+			continue // all rules evaluated by lifecycle worker at scan time
+		}
+
+		var rulePrefix string
+		switch {
+		case rule.Filter.andSet:
+			rulePrefix = rule.Filter.And.Prefix.val
+		case rule.Filter.Prefix.set:
+			rulePrefix = rule.Filter.Prefix.val
+		case rule.Prefix.set:
+			rulePrefix = rule.Prefix.val
+		}
+
+		// Only create filer.conf TTL entries for simple Expiration.Days rules
+		// with prefix-only filters (the fast path handled by RocksDB compaction
+		// filter). Rules with tag or size filters must be evaluated at scan time
+		// by the lifecycle worker, because TTL applies to all objects under the
+		// prefix regardless of tags or size.
 		if rule.Expiration.Days == 0 {
 			continue
 		}
+		hasTagOrSizeFilter := rule.Filter.tagSet ||
+			rule.Filter.ObjectSizeGreaterThan > 0 || rule.Filter.ObjectSizeLessThan > 0 ||
+			(rule.Filter.andSet && (len(rule.Filter.And.Tags) > 0 ||
+				rule.Filter.And.ObjectSizeGreaterThan > 0 || rule.Filter.And.ObjectSizeLessThan > 0))
+		if hasTagOrSizeFilter {
+			continue // evaluated by lifecycle worker at scan time
+		}
 		locationPrefix := fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, bucket, rulePrefix)
 		locConf := &filer_pb.FilerConf_PathConf{
 			LocationPrefix:    locationPrefix,

weed/s3api/s3api_bucket_lifecycle_response_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -124,3 +125,39 @@ func (f failingReadCloser) Read(_ []byte) (int, error) {
 func (f failingReadCloser) Close() error {
 	return nil
 }
+
+// TestShouldSkipTTLFastPathForVersionedBuckets verifies the versioning guard
+// logic that PutBucketLifecycleConfigurationHandler uses to decide whether
+// to create filer.conf TTL entries. On AWS S3, Expiration.Days on a versioned
+// bucket creates a delete marker — it does not delete data. TTL volumes
+// would destroy all versions indiscriminately, so the lifecycle worker must
+// handle versioned buckets at scan time instead. (issue #8757)
+//
+// Note: an integration test that invokes PutBucketLifecycleConfigurationHandler
+// directly is not feasible here because the handler requires filer gRPC
+// connectivity (ReadFilerConfFromFilers, WithFilerClient) before it reaches
+// the versioning check. The lifecycle worker integration tests in
+// weed/plugin/worker/lifecycle/ cover the end-to-end versioned-bucket behavior.
+func TestShouldSkipTTLFastPathForVersionedBuckets(t *testing.T) {
+	tests := []struct {
+		name       string
+		versioning string
+		expectSkip bool
+	}{
+		{"versioning_enabled_skips_ttl", s3_constants.VersioningEnabled, true},
+		{"versioning_suspended_skips_ttl", s3_constants.VersioningSuspended, true},
+		{"unversioned_allows_ttl", "", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// This mirrors the guard in PutBucketLifecycleConfigurationHandler:
+			//   isVersioned := versioningErr != s3err.ErrNone ||
+			//       bucketVersioning == s3_constants.VersioningEnabled ||
+			//       bucketVersioning == s3_constants.VersioningSuspended
+			// When isVersioned is true, the TTL fast-path is skipped entirely.
+			isVersioned := tt.versioning == s3_constants.VersioningEnabled ||
+				tt.versioning == s3_constants.VersioningSuspended
+			assert.Equal(t, tt.expectSkip, isVersioned)
+		})
+	}
+}

weed/s3api/s3api_key_rotation.go

@@ -1,275 +1,12 @@
 package s3api
 
 import (
-	"bytes"
-	"crypto/rand"
-	"fmt"
-	"io"
 	"net/http"
 
-	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
 
-// rotateSSECKey handles SSE-C key rotation for same-object copies
-func (s3a *S3ApiServer) rotateSSECKey(entry *filer_pb.Entry, r *http.Request) ([]*filer_pb.FileChunk, error) {
-	// Parse source and destination SSE-C keys
-	sourceKey, err := ParseSSECCopySourceHeaders(r)
-	if err != nil {
-		return nil, fmt.Errorf("parse SSE-C copy source headers: %w", err)
-	}
-
-	destKey, err := ParseSSECHeaders(r)
-	if err != nil {
-		return nil, fmt.Errorf("parse SSE-C destination headers: %w", err)
-	}
-
-	// Validate that we have both keys
-	if sourceKey == nil {
-		return nil, fmt.Errorf("source SSE-C key required for key rotation")
-	}
-
-	if destKey == nil {
-		return nil, fmt.Errorf("destination SSE-C key required for key rotation")
-	}
-
-	// Check if keys are actually different
-	if sourceKey.KeyMD5 == destKey.KeyMD5 {
-		glog.V(2).Infof("SSE-C key rotation: keys are identical, using direct copy")
-		return entry.GetChunks(), nil
-	}
-
-	glog.V(2).Infof("SSE-C key rotation: rotating from key %s to key %s",
-		sourceKey.KeyMD5[:8], destKey.KeyMD5[:8])
-
-	// For SSE-C key rotation, we need to re-encrypt all chunks
-	// This cannot be a metadata-only operation because the encryption key changes
-	return s3a.rotateSSECChunks(entry, sourceKey, destKey)
-}
-
-// rotateSSEKMSKey handles SSE-KMS key rotation for same-object copies
-func (s3a *S3ApiServer) rotateSSEKMSKey(entry *filer_pb.Entry, r *http.Request) ([]*filer_pb.FileChunk, error) {
-	// Get source and destination key IDs
-	srcKeyID, srcEncrypted := GetSourceSSEKMSInfo(entry.Extended)
-	if !srcEncrypted {
-		return nil, fmt.Errorf("source object is not SSE-KMS encrypted")
-	}
-
-	dstKeyID := r.Header.Get(s3_constants.AmzServerSideEncryptionAwsKmsKeyId)
-	if dstKeyID == "" {
-		// Use default key if not specified
-		dstKeyID = "default"
-	}
-
-	// Check if keys are actually different
-	if srcKeyID == dstKeyID {
-		glog.V(2).Infof("SSE-KMS key rotation: keys are identical, using direct copy")
-		return entry.GetChunks(), nil
-	}
-
-	glog.V(2).Infof("SSE-KMS key rotation: rotating from key %s to key %s", srcKeyID, dstKeyID)
-
-	// For SSE-KMS, we can potentially do metadata-only rotation
-	// if the KMS service supports key aliasing and the data encryption key can be re-wrapped
-	if s3a.CanDoMetadataOnlyKMSRotation(srcKeyID, dstKeyID) {
-		return s3a.rotateSSEKMSMetadataOnly(entry, srcKeyID, dstKeyID)
-	}
-
-	// Fallback to full re-encryption
-	return s3a.rotateSSEKMSChunks(entry, srcKeyID, dstKeyID, r)
-}
-
-// CanDoMetadataOnlyKMSRotation determines if KMS key rotation can be done metadata-only
-func (s3a *S3ApiServer) CanDoMetadataOnlyKMSRotation(srcKeyID, dstKeyID string) bool {
-	// For now, we'll be conservative and always re-encrypt
-	// In a full implementation, this would check if:
-	// 1. Both keys are in the same KMS instance
-	// 2. The KMS supports key re-wrapping
-	// 3. The user has permissions for both keys
-	return false
-}
-
-// rotateSSEKMSMetadataOnly performs metadata-only SSE-KMS key rotation
-func (s3a *S3ApiServer) rotateSSEKMSMetadataOnly(entry *filer_pb.Entry, srcKeyID, dstKeyID string) ([]*filer_pb.FileChunk, error) {
-	// This would re-wrap the data encryption key with the new KMS key
-	// For now, return an error since we don't support this yet
-	return nil, fmt.Errorf("metadata-only KMS key rotation not yet implemented")
-}
-
-// rotateSSECChunks re-encrypts all chunks with new SSE-C key
-func (s3a *S3ApiServer) rotateSSECChunks(entry *filer_pb.Entry, sourceKey, destKey *SSECustomerKey) ([]*filer_pb.FileChunk, error) {
-	// Get IV from entry metadata
-	iv, err := GetSSECIVFromMetadata(entry.Extended)
-	if err != nil {
-		return nil, fmt.Errorf("get SSE-C IV from metadata: %w", err)
-	}
-
-	var rotatedChunks []*filer_pb.FileChunk
-
-	for _, chunk := range entry.GetChunks() {
-		rotatedChunk, err := s3a.rotateSSECChunk(chunk, sourceKey, destKey, iv)
-		if err != nil {
-			return nil, fmt.Errorf("rotate SSE-C chunk: %w", err)
-		}
-		rotatedChunks = append(rotatedChunks, rotatedChunk)
-	}
-
-	// Generate new IV for the destination and store it in entry metadata
-	newIV := make([]byte, s3_constants.AESBlockSize)
-	if _, err := io.ReadFull(rand.Reader, newIV); err != nil {
-		return nil, fmt.Errorf("generate new IV: %w", err)
-	}
-
-	// Update entry metadata with new IV and SSE-C headers
-	if entry.Extended == nil {
-		entry.Extended = make(map[string][]byte)
-	}
-	StoreSSECIVInMetadata(entry.Extended, newIV)
-	entry.Extended[s3_constants.AmzServerSideEncryptionCustomerAlgorithm] = []byte("AES256")
-	entry.Extended[s3_constants.AmzServerSideEncryptionCustomerKeyMD5] = []byte(destKey.KeyMD5)
-
-	return rotatedChunks, nil
-}
-
-// rotateSSEKMSChunks re-encrypts all chunks with new SSE-KMS key
-func (s3a *S3ApiServer) rotateSSEKMSChunks(entry *filer_pb.Entry, srcKeyID, dstKeyID string, r *http.Request) ([]*filer_pb.FileChunk, error) {
-	var rotatedChunks []*filer_pb.FileChunk
-
-	// Parse encryption context and bucket key settings
-	_, encryptionContext, bucketKeyEnabled, err := ParseSSEKMSCopyHeaders(r)
-	if err != nil {
-		return nil, fmt.Errorf("parse SSE-KMS copy headers: %w", err)
-	}
-
-	for _, chunk := range entry.GetChunks() {
-		rotatedChunk, err := s3a.rotateSSEKMSChunk(chunk, srcKeyID, dstKeyID, encryptionContext, bucketKeyEnabled)
-		if err != nil {
-			return nil, fmt.Errorf("rotate SSE-KMS chunk: %w", err)
-		}
-		rotatedChunks = append(rotatedChunks, rotatedChunk)
-	}
-
-	return rotatedChunks, nil
-}
-
-// rotateSSECChunk rotates a single SSE-C encrypted chunk
-func (s3a *S3ApiServer) rotateSSECChunk(chunk *filer_pb.FileChunk, sourceKey, destKey *SSECustomerKey, iv []byte) (*filer_pb.FileChunk, error) {
-	// Create new chunk with same properties
-	newChunk := &filer_pb.FileChunk{
-		Offset:       chunk.Offset,
-		Size:         chunk.Size,
-		ModifiedTsNs: chunk.ModifiedTsNs,
-		ETag:         chunk.ETag,
-	}
-
-	// Assign new volume for the rotated chunk
-	assignResult, err := s3a.assignNewVolume("")
-	if err != nil {
-		return nil, fmt.Errorf("assign new volume: %w", err)
-	}
-
-	// Set file ID on new chunk
-	if err := s3a.setChunkFileId(newChunk, assignResult); err != nil {
-		return nil, err
-	}
-
-	// Get source chunk data
-	fileId := chunk.GetFileIdString()
-	srcUrl, err := s3a.lookupVolumeUrl(fileId)
-	if err != nil {
-		return nil, fmt.Errorf("lookup source volume: %w", err)
-	}
-
-	// Download encrypted data
-	encryptedData, err := s3a.downloadChunkData(srcUrl, fileId, 0, int64(chunk.Size), chunk.CipherKey)
-	if err != nil {
-		return nil, fmt.Errorf("download chunk data: %w", err)
-	}
-
-	// Decrypt with source key using provided IV
-	decryptedReader, err := CreateSSECDecryptedReader(bytes.NewReader(encryptedData), sourceKey, iv)
-	if err != nil {
-		return nil, fmt.Errorf("create decrypted reader: %w", err)
-	}
-
-	decryptedData, err := io.ReadAll(decryptedReader)
-	if err != nil {
-		return nil, fmt.Errorf("decrypt data: %w", err)
-	}
-
-	// Re-encrypt with destination key
-	encryptedReader, _, err := CreateSSECEncryptedReader(bytes.NewReader(decryptedData), destKey)
-	if err != nil {
-		return nil, fmt.Errorf("create encrypted reader: %w", err)
-	}
-
-	// Note: IV will be handled at the entry level by the calling function
-
-	reencryptedData, err := io.ReadAll(encryptedReader)
-	if err != nil {
-		return nil, fmt.Errorf("re-encrypt data: %w", err)
-	}
-
-	// Update chunk size to include new IV
-	newChunk.Size = uint64(len(reencryptedData))
-
-	// Upload re-encrypted data
-	if err := s3a.uploadChunkData(reencryptedData, assignResult, false); err != nil {
-		return nil, fmt.Errorf("upload re-encrypted data: %w", err)
-	}
-
-	return newChunk, nil
-}
-
-// rotateSSEKMSChunk rotates a single SSE-KMS encrypted chunk
-func (s3a *S3ApiServer) rotateSSEKMSChunk(chunk *filer_pb.FileChunk, srcKeyID, dstKeyID string, encryptionContext map[string]string, bucketKeyEnabled bool) (*filer_pb.FileChunk, error) {
-	// Create new chunk with same properties
-	newChunk := &filer_pb.FileChunk{
-		Offset:       chunk.Offset,
-		Size:         chunk.Size,
-		ModifiedTsNs: chunk.ModifiedTsNs,
-		ETag:         chunk.ETag,
-	}
-
-	// Assign new volume for the rotated chunk
-	assignResult, err := s3a.assignNewVolume("")
-	if err != nil {
-		return nil, fmt.Errorf("assign new volume: %w", err)
-	}
-
-	// Set file ID on new chunk
-	if err := s3a.setChunkFileId(newChunk, assignResult); err != nil {
-		return nil, err
-	}
-
-	// Get source chunk data
-	fileId := chunk.GetFileIdString()
-	srcUrl, err := s3a.lookupVolumeUrl(fileId)
-	if err != nil {
-		return nil, fmt.Errorf("lookup source volume: %w", err)
-	}
-
-	// Download data (this would be encrypted with the old KMS key)
-	chunkData, err := s3a.downloadChunkData(srcUrl, fileId, 0, int64(chunk.Size), chunk.CipherKey)
-	if err != nil {
-		return nil, fmt.Errorf("download chunk data: %w", err)
-	}
-
-	// For now, we'll just re-upload the data as-is
-	// In a full implementation, this would:
-	// 1. Decrypt with old KMS key
-	// 2. Re-encrypt with new KMS key
-	// 3. Update metadata accordingly
-
-	// Upload data with new key (placeholder implementation)
-	if err := s3a.uploadChunkData(chunkData, assignResult, false); err != nil {
-		return nil, fmt.Errorf("upload rotated data: %w", err)
-	}
-
-	return newChunk, nil
-}
-
 // IsSameObjectCopy determines if this is a same-object copy operation
 func IsSameObjectCopy(r *http.Request, srcBucket, srcObject, dstBucket, dstObject string) bool {
 	return srcBucket == dstBucket && srcObject == dstObject

weed/s3api/s3api_object_handlers.go

@@ -244,11 +244,6 @@ func (e *StreamError) Unwrap() error {
 	return e.Err
 }
 
-// newStreamError creates a StreamError for cases where response hasn't been written yet
-func newStreamError(err error) *StreamError {
-	return &StreamError{Err: err, ResponseWritten: false}
-}
-
 // newStreamErrorWithResponse creates a StreamError for cases where response was already written
 func newStreamErrorWithResponse(err error) *StreamError {
 	return &StreamError{Err: err, ResponseWritten: true}

weed/s3api/s3api_object_handlers_copy.go

@@ -2099,18 +2099,6 @@ func (s3a *S3ApiServer) copyCrossEncryptionChunk(chunk *filer_pb.FileChunk, sour
 	return dstChunk, nil
 }
 
-// getEncryptionTypeString returns a string representation of encryption type for logging
-func (s3a *S3ApiServer) getEncryptionTypeString(isSSEC, isSSEKMS, isSSES3 bool) string {
-	if isSSEC {
-		return s3_constants.SSETypeC
-	} else if isSSEKMS {
-		return s3_constants.SSETypeKMS
-	} else if isSSES3 {
-		return s3_constants.SSETypeS3
-	}
-	return "Plain"
-}
-
 // copyChunksWithSSEC handles SSE-C aware copying with smart fast/slow path selection
 // Returns chunks and destination metadata that should be applied to the destination entry
 func (s3a *S3ApiServer) copyChunksWithSSEC(entry *filer_pb.Entry, r *http.Request) ([]*filer_pb.FileChunk, map[string][]byte, error) {

weed/s3api/s3api_object_handlers_list.go

@@ -762,60 +762,6 @@ func getListObjectsV1Args(values url.Values) (prefix, marker, delimiter string,
 	return
 }
 
-func (s3a *S3ApiServer) ensureDirectoryAllEmpty(filerClient filer_pb.SeaweedFilerClient, parentDir, name string) (isEmpty bool, err error) {
-	// println("+ ensureDirectoryAllEmpty", dir, name)
-	glog.V(4).Infof("+ isEmpty %s/%s", parentDir, name)
-	defer glog.V(4).Infof("- isEmpty %s/%s %v", parentDir, name, isEmpty)
-	var fileCounter int
-	var subDirs []string
-	currentDir := parentDir + "/" + name
-	var startFrom string
-	var isExhausted bool
-	var foundEntry bool
-	for fileCounter == 0 && !isExhausted && err == nil {
-		err = filer_pb.SeaweedList(context.Background(), filerClient, currentDir, "", func(entry *filer_pb.Entry, isLast bool) error {
-			foundEntry = true
-			if entry.IsOlderDir() {
-				subDirs = append(subDirs, entry.Name)
-			} else {
-				fileCounter++
-			}
-			startFrom = entry.Name
-			isExhausted = isExhausted || isLast
-			glog.V(4).Infof("    * %s/%s isLast: %t", currentDir, startFrom, isLast)
-			return nil
-		}, startFrom, false, 8)
-		if !foundEntry {
-			break
-		}
-	}
-
-	if err != nil {
-		return false, err
-	}
-
-	if fileCounter > 0 {
-		return false, nil
-	}
-
-	for _, subDir := range subDirs {
-		isSubEmpty, subErr := s3a.ensureDirectoryAllEmpty(filerClient, currentDir, subDir)
-		if subErr != nil {
-			return false, subErr
-		}
-		if !isSubEmpty {
-			return false, nil
-		}
-	}
-
-	glog.V(1).Infof("deleting empty folder %s", currentDir)
-	if err = doDeleteEntry(filerClient, parentDir, name, true, false); err != nil {
-		return
-	}
-
-	return true, nil
-}
-
 // compareWithDelimiter compares two strings for sorting, treating the delimiter character
 // as having lower precedence than other characters to match AWS S3 behavior.
 // For example, with delimiter '/', 'foo/' should come before 'foo+1/' even though '+' < '/' in ASCII.

weed/s3api/s3api_object_retention.go

@@ -463,24 +463,6 @@ func (s3a *S3ApiServer) setObjectLegalHold(bucket, object, versionId string, leg
 // PROTECTION ENFORCEMENT
 // ====================================================================
 
-// isObjectRetentionActive checks if object has active retention
-func (s3a *S3ApiServer) isObjectRetentionActive(bucket, object, versionId string) (bool, error) {
-	retention, err := s3a.getObjectRetention(bucket, object, versionId)
-	if err != nil {
-		// If no retention found, object is not under retention
-		if errors.Is(err, ErrNoRetentionConfiguration) {
-			return false, nil
-		}
-		return false, err
-	}
-
-	if retention.RetainUntilDate != nil && retention.RetainUntilDate.After(time.Now()) {
-		return true, nil
-	}
-
-	return false, nil
-}
-
 // getRetentionFromEntry extracts retention configuration from filer entry
 func (s3a *S3ApiServer) getRetentionFromEntry(entry *filer_pb.Entry) (*ObjectRetention, bool, error) {
 	if entry.Extended == nil {

weed/s3api/s3api_object_versioning.go

@@ -72,16 +72,6 @@ func setCachedListMetadata(versionsEntry, versionEntry *filer_pb.Entry) {
 	}
 }
 
-// clearCachedListMetadata removes all cached list metadata from the .versions directory entry
-func clearCachedListMetadata(extended map[string][]byte) {
-	if extended == nil {
-		return
-	}
-	delete(extended, s3_constants.ExtLatestVersionIdKey)
-	delete(extended, s3_constants.ExtLatestVersionFileNameKey)
-	clearCachedVersionMetadata(extended)
-}
-
 // S3ListObjectVersionsResult - Custom struct for S3 list-object-versions response.
 // This avoids conflicts with the XSD generated ListVersionsResult struct.
 //

weed/s3api/s3api_policy.go

@@ -22,13 +22,16 @@ type Lifecycle struct {
 
 // Rule - a rule for lifecycle configuration.
 type Rule struct {
-	XMLName    xml.Name   `xml:"Rule"`
-	ID         string     `xml:"ID,omitempty"`
-	Status     ruleStatus `xml:"Status"`
-	Filter     Filter     `xml:"Filter,omitempty"`
-	Prefix     Prefix     `xml:"Prefix,omitempty"`
-	Expiration Expiration `xml:"Expiration,omitempty"`
-	Transition Transition `xml:"Transition,omitempty"`
+	XMLName                        xml.Name                       `xml:"Rule"`
+	ID                             string                         `xml:"ID,omitempty"`
+	Status                         ruleStatus                     `xml:"Status"`
+	Filter                         Filter                         `xml:"Filter,omitempty"`
+	Prefix                         Prefix                         `xml:"Prefix,omitempty"`
+	Expiration                     Expiration                     `xml:"Expiration,omitempty"`
+	Transition                     Transition                     `xml:"Transition,omitempty"`
+	NoncurrentVersionExpiration    NoncurrentVersionExpiration    `xml:"NoncurrentVersionExpiration,omitempty"`
+	NoncurrentVersionTransition    NoncurrentVersionTransition    `xml:"NoncurrentVersionTransition,omitempty"`
+	AbortIncompleteMultipartUpload AbortIncompleteMultipartUpload `xml:"AbortIncompleteMultipartUpload,omitempty"`
 }
 
 // Filter - a filter for a lifecycle configuration Rule.
@@ -43,6 +46,9 @@ type Filter struct {
 
 	Tag    Tag
 	tagSet bool
+
+	ObjectSizeGreaterThan int64
+	ObjectSizeLessThan    int64
 }
 
 // Prefix holds the prefix xml tag in <Rule> and <Filter>
@@ -80,17 +86,83 @@ func (f Filter) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
 	if err := e.EncodeToken(start); err != nil {
 		return err
 	}
-	if err := e.EncodeElement(f.Prefix, xml.StartElement{Name: xml.Name{Local: "Prefix"}}); err != nil {
-		return err
+	if f.andSet {
+		if err := e.EncodeElement(f.And, xml.StartElement{Name: xml.Name{Local: "And"}}); err != nil {
+			return err
+		}
+	} else if f.tagSet {
+		if err := e.EncodeElement(f.Tag, xml.StartElement{Name: xml.Name{Local: "Tag"}}); err != nil {
+			return err
+		}
+	} else {
+		if err := e.EncodeElement(f.Prefix, xml.StartElement{Name: xml.Name{Local: "Prefix"}}); err != nil {
+			return err
+		}
+	}
+	if f.ObjectSizeGreaterThan > 0 {
+		if err := e.EncodeElement(f.ObjectSizeGreaterThan, xml.StartElement{Name: xml.Name{Local: "ObjectSizeGreaterThan"}}); err != nil {
+			return err
+		}
+	}
+	if f.ObjectSizeLessThan > 0 {
+		if err := e.EncodeElement(f.ObjectSizeLessThan, xml.StartElement{Name: xml.Name{Local: "ObjectSizeLessThan"}}); err != nil {
+			return err
+		}
 	}
 	return e.EncodeToken(xml.EndElement{Name: start.Name})
 }
 
+// UnmarshalXML decodes Filter from XML, handling all child elements.
+func (f *Filter) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	f.set = true
+	for {
+		tok, err := d.Token()
+		if err != nil {
+			return err
+		}
+		switch t := tok.(type) {
+		case xml.StartElement:
+			switch t.Name.Local {
+			case "Prefix":
+				if err := d.DecodeElement(&f.Prefix, &t); err != nil {
+					return err
+				}
+			case "Tag":
+				f.tagSet = true
+				if err := d.DecodeElement(&f.Tag, &t); err != nil {
+					return err
+				}
+			case "And":
+				f.andSet = true
+				if err := d.DecodeElement(&f.And, &t); err != nil {
+					return err
+				}
+			case "ObjectSizeGreaterThan":
+				if err := d.DecodeElement(&f.ObjectSizeGreaterThan, &t); err != nil {
+					return err
+				}
+			case "ObjectSizeLessThan":
+				if err := d.DecodeElement(&f.ObjectSizeLessThan, &t); err != nil {
+					return err
+				}
+			default:
+				if err := d.Skip(); err != nil {
+					return err
+				}
+			}
+		case xml.EndElement:
+			return nil
+		}
+	}
+}
+
 // And - a tag to combine a prefix and multiple tags for lifecycle configuration rule.
 type And struct {
-	XMLName xml.Name `xml:"And"`
-	Prefix  Prefix   `xml:"Prefix,omitempty"`
-	Tags    []Tag    `xml:"Tag,omitempty"`
+	XMLName               xml.Name `xml:"And"`
+	Prefix                Prefix   `xml:"Prefix,omitempty"`
+	Tags                  []Tag    `xml:"Tag,omitempty"`
+	ObjectSizeGreaterThan int64    `xml:"ObjectSizeGreaterThan,omitempty"`
+	ObjectSizeLessThan    int64    `xml:"ObjectSizeLessThan,omitempty"`
 }
 
 // Expiration - expiration actions for a rule in lifecycle configuration.
@@ -112,6 +184,17 @@ func (e Expiration) MarshalXML(enc *xml.Encoder, startElement xml.StartElement)
 	return enc.EncodeElement(expirationWrapper(e), startElement)
 }
 
+func (e *Expiration) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	type wrapper Expiration
+	var w wrapper
+	if err := d.DecodeElement(&w, &start); err != nil {
+		return err
+	}
+	*e = Expiration(w)
+	e.set = true
+	return nil
+}
+
 // ExpireDeleteMarker represents value of ExpiredObjectDeleteMarker field in Expiration XML element.
 type ExpireDeleteMarker struct {
 	val bool
@@ -126,6 +209,15 @@ func (b ExpireDeleteMarker) MarshalXML(e *xml.Encoder, startElement xml.StartEle
 	return e.EncodeElement(b.val, startElement)
 }
 
+func (b *ExpireDeleteMarker) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	var v bool
+	if err := d.DecodeElement(&v, &start); err != nil {
+		return err
+	}
+	*b = ExpireDeleteMarker{val: v, set: true}
+	return nil
+}
+
 // ExpirationDate is a embedded type containing time.Time to unmarshal
 // Date in Expiration
 type ExpirationDate struct {
@@ -160,5 +252,102 @@ func (t Transition) MarshalXML(enc *xml.Encoder, start xml.StartElement) error {
 	return enc.EncodeElement(transitionWrapper(t), start)
 }
 
+func (t *Transition) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	type wrapper Transition
+	var w wrapper
+	if err := d.DecodeElement(&w, &start); err != nil {
+		return err
+	}
+	*t = Transition(w)
+	t.set = true
+	return nil
+}
+
 // TransitionDays is a type alias to unmarshal Days in Transition
 type TransitionDays int
+
+// NoncurrentVersionExpiration - expiration actions for non-current object versions.
+type NoncurrentVersionExpiration struct {
+	XMLName                 xml.Name `xml:"NoncurrentVersionExpiration"`
+	NoncurrentDays          int      `xml:"NoncurrentDays,omitempty"`
+	NewerNoncurrentVersions int      `xml:"NewerNoncurrentVersions,omitempty"`
+
+	set bool
+}
+
+// MarshalXML encodes NoncurrentVersionExpiration field into an XML form.
+func (n NoncurrentVersionExpiration) MarshalXML(enc *xml.Encoder, start xml.StartElement) error {
+	if !n.set {
+		return nil
+	}
+	type wrapper NoncurrentVersionExpiration
+	return enc.EncodeElement(wrapper(n), start)
+}
+
+func (n *NoncurrentVersionExpiration) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	type wrapper NoncurrentVersionExpiration
+	var w wrapper
+	if err := d.DecodeElement(&w, &start); err != nil {
+		return err
+	}
+	*n = NoncurrentVersionExpiration(w)
+	n.set = true
+	return nil
+}
+
+// NoncurrentVersionTransition - transition actions for non-current object versions.
+type NoncurrentVersionTransition struct {
+	XMLName        xml.Name `xml:"NoncurrentVersionTransition"`
+	NoncurrentDays int      `xml:"NoncurrentDays,omitempty"`
+	StorageClass   string   `xml:"StorageClass,omitempty"`
+
+	set bool
+}
+
+// MarshalXML encodes NoncurrentVersionTransition field into an XML form.
+func (n NoncurrentVersionTransition) MarshalXML(enc *xml.Encoder, start xml.StartElement) error {
+	if !n.set {
+		return nil
+	}
+	type wrapper NoncurrentVersionTransition
+	return enc.EncodeElement(wrapper(n), start)
+}
+
+func (n *NoncurrentVersionTransition) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	type wrapper NoncurrentVersionTransition
+	var w wrapper
+	if err := d.DecodeElement(&w, &start); err != nil {
+		return err
+	}
+	*n = NoncurrentVersionTransition(w)
+	n.set = true
+	return nil
+}
+
+// AbortIncompleteMultipartUpload - abort action for incomplete multipart uploads.
+type AbortIncompleteMultipartUpload struct {
+	XMLName             xml.Name `xml:"AbortIncompleteMultipartUpload"`
+	DaysAfterInitiation int      `xml:"DaysAfterInitiation,omitempty"`
+
+	set bool
+}
+
+// MarshalXML encodes AbortIncompleteMultipartUpload field into an XML form.
+func (a AbortIncompleteMultipartUpload) MarshalXML(enc *xml.Encoder, start xml.StartElement) error {
+	if !a.set {
+		return nil
+	}
+	type wrapper AbortIncompleteMultipartUpload
+	return enc.EncodeElement(wrapper(a), start)
+}
+
+func (a *AbortIncompleteMultipartUpload) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	type wrapper AbortIncompleteMultipartUpload
+	var w wrapper
+	if err := d.DecodeElement(&w, &start); err != nil {
+		return err
+	}
+	*a = AbortIncompleteMultipartUpload(w)
+	a.set = true
+	return nil
+}

weed/s3api/s3api_policy_test.go

@@ -0,0 +1,276 @@
+package s3api
+
+import (
+	"encoding/xml"
+	"strings"
+	"testing"
+)
+
+func TestLifecycleXMLRoundTrip_NoncurrentVersionExpiration(t *testing.T) {
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>expire-noncurrent</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <NoncurrentVersionExpiration>
+      <NoncurrentDays>30</NoncurrentDays>
+      <NewerNoncurrentVersions>2</NewerNoncurrentVersions>
+    </NoncurrentVersionExpiration>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	if len(lc.Rules) != 1 {
+		t.Fatalf("expected 1 rule, got %d", len(lc.Rules))
+	}
+	rule := lc.Rules[0]
+	if rule.ID != "expire-noncurrent" {
+		t.Errorf("expected ID 'expire-noncurrent', got %q", rule.ID)
+	}
+	if rule.NoncurrentVersionExpiration.NoncurrentDays != 30 {
+		t.Errorf("expected NoncurrentDays=30, got %d", rule.NoncurrentVersionExpiration.NoncurrentDays)
+	}
+	if rule.NoncurrentVersionExpiration.NewerNoncurrentVersions != 2 {
+		t.Errorf("expected NewerNoncurrentVersions=2, got %d", rule.NoncurrentVersionExpiration.NewerNoncurrentVersions)
+	}
+
+	// Re-marshal and verify it round-trips.
+	out, err := xml.Marshal(lc)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	s := string(out)
+	if !strings.Contains(s, "<NoncurrentDays>30</NoncurrentDays>") {
+		t.Errorf("marshaled XML missing NoncurrentDays: %s", s)
+	}
+	if !strings.Contains(s, "<NewerNoncurrentVersions>2</NewerNoncurrentVersions>") {
+		t.Errorf("marshaled XML missing NewerNoncurrentVersions: %s", s)
+	}
+}
+
+func TestLifecycleXMLRoundTrip_AbortIncompleteMultipartUpload(t *testing.T) {
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>abort-mpu</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <AbortIncompleteMultipartUpload>
+      <DaysAfterInitiation>7</DaysAfterInitiation>
+    </AbortIncompleteMultipartUpload>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	rule := lc.Rules[0]
+	if rule.AbortIncompleteMultipartUpload.DaysAfterInitiation != 7 {
+		t.Errorf("expected DaysAfterInitiation=7, got %d", rule.AbortIncompleteMultipartUpload.DaysAfterInitiation)
+	}
+
+	out, err := xml.Marshal(lc)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	if !strings.Contains(string(out), "<DaysAfterInitiation>7</DaysAfterInitiation>") {
+		t.Errorf("marshaled XML missing DaysAfterInitiation: %s", string(out))
+	}
+}
+
+func TestLifecycleXMLRoundTrip_FilterWithTag(t *testing.T) {
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>tag-filter</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <Tag><Key>env</Key><Value>dev</Value></Tag>
+    </Filter>
+    <Expiration><Days>7</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	rule := lc.Rules[0]
+	if !rule.Filter.tagSet {
+		t.Error("expected Filter.tagSet to be true")
+	}
+	if rule.Filter.Tag.Key != "env" || rule.Filter.Tag.Value != "dev" {
+		t.Errorf("expected Tag{env:dev}, got Tag{%s:%s}", rule.Filter.Tag.Key, rule.Filter.Tag.Value)
+	}
+}
+
+func TestLifecycleXMLRoundTrip_FilterWithAnd(t *testing.T) {
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>and-filter</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <And>
+        <Prefix>logs/</Prefix>
+        <Tag><Key>env</Key><Value>dev</Value></Tag>
+        <Tag><Key>tier</Key><Value>hot</Value></Tag>
+        <ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan>
+        <ObjectSizeLessThan>1048576</ObjectSizeLessThan>
+      </And>
+    </Filter>
+    <Expiration><Days>7</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	rule := lc.Rules[0]
+	if !rule.Filter.andSet {
+		t.Error("expected Filter.andSet to be true")
+	}
+	if rule.Filter.And.Prefix.String() != "logs/" {
+		t.Errorf("expected And.Prefix='logs/', got %q", rule.Filter.And.Prefix.String())
+	}
+	if len(rule.Filter.And.Tags) != 2 {
+		t.Fatalf("expected 2 And tags, got %d", len(rule.Filter.And.Tags))
+	}
+	if rule.Filter.And.ObjectSizeGreaterThan != 1024 {
+		t.Errorf("expected ObjectSizeGreaterThan=1024, got %d", rule.Filter.And.ObjectSizeGreaterThan)
+	}
+	if rule.Filter.And.ObjectSizeLessThan != 1048576 {
+		t.Errorf("expected ObjectSizeLessThan=1048576, got %d", rule.Filter.And.ObjectSizeLessThan)
+	}
+}
+
+func TestLifecycleXMLRoundTrip_FilterWithSizeOnly(t *testing.T) {
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>size-filter</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <ObjectSizeGreaterThan>512</ObjectSizeGreaterThan>
+      <ObjectSizeLessThan>10485760</ObjectSizeLessThan>
+    </Filter>
+    <Expiration><Days>30</Days></Expiration>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	rule := lc.Rules[0]
+	if rule.Filter.ObjectSizeGreaterThan != 512 {
+		t.Errorf("expected ObjectSizeGreaterThan=512, got %d", rule.Filter.ObjectSizeGreaterThan)
+	}
+	if rule.Filter.ObjectSizeLessThan != 10485760 {
+		t.Errorf("expected ObjectSizeLessThan=10485760, got %d", rule.Filter.ObjectSizeLessThan)
+	}
+}
+
+func TestLifecycleXML_TransitionSetFlag(t *testing.T) {
+	// Verify that Transition.set is true after unmarshaling.
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>transition</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <Transition>
+      <Days>30</Days>
+      <StorageClass>GLACIER</StorageClass>
+    </Transition>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if !lc.Rules[0].Transition.set {
+		t.Error("expected Transition.set=true after unmarshal")
+	}
+}
+
+func TestLifecycleXML_NoncurrentVersionTransitionSetFlag(t *testing.T) {
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>nv-transition</ID>
+    <Status>Enabled</Status>
+    <Filter><Prefix></Prefix></Filter>
+    <NoncurrentVersionTransition>
+      <NoncurrentDays>60</NoncurrentDays>
+      <StorageClass>GLACIER</StorageClass>
+    </NoncurrentVersionTransition>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if !lc.Rules[0].NoncurrentVersionTransition.set {
+		t.Error("expected NoncurrentVersionTransition.set=true after unmarshal")
+	}
+}
+
+func TestLifecycleXMLRoundTrip_CompleteRule(t *testing.T) {
+	// A complete lifecycle config similar to what Terraform sends.
+	input := `<LifecycleConfiguration>
+  <Rule>
+    <ID>rotation</ID>
+    <Filter><Prefix></Prefix></Filter>
+    <Status>Enabled</Status>
+    <Expiration><Days>30</Days></Expiration>
+    <NoncurrentVersionExpiration>
+      <NoncurrentDays>1</NoncurrentDays>
+    </NoncurrentVersionExpiration>
+    <AbortIncompleteMultipartUpload>
+      <DaysAfterInitiation>1</DaysAfterInitiation>
+    </AbortIncompleteMultipartUpload>
+  </Rule>
+</LifecycleConfiguration>`
+
+	var lc Lifecycle
+	if err := xml.Unmarshal([]byte(input), &lc); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	rule := lc.Rules[0]
+	if rule.ID != "rotation" {
+		t.Errorf("expected ID 'rotation', got %q", rule.ID)
+	}
+	if rule.Expiration.Days != 30 {
+		t.Errorf("expected Expiration.Days=30, got %d", rule.Expiration.Days)
+	}
+	if rule.NoncurrentVersionExpiration.NoncurrentDays != 1 {
+		t.Errorf("expected NoncurrentDays=1, got %d", rule.NoncurrentVersionExpiration.NoncurrentDays)
+	}
+	if rule.AbortIncompleteMultipartUpload.DaysAfterInitiation != 1 {
+		t.Errorf("expected DaysAfterInitiation=1, got %d", rule.AbortIncompleteMultipartUpload.DaysAfterInitiation)
+	}
+
+	// Re-marshal and verify all fields survive.
+	out, err := xml.Marshal(lc)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	s := string(out)
+	for _, expected := range []string{
+		"<Days>30</Days>",
+		"<NoncurrentDays>1</NoncurrentDays>",
+		"<DaysAfterInitiation>1</DaysAfterInitiation>",
+	} {
+		if !strings.Contains(s, expected) {
+			t.Errorf("marshaled XML missing %q: %s", expected, s)
+		}
+	}
+}

weed/s3api/s3api_put_handlers.go

@@ -23,20 +23,6 @@ type PutToFilerEncryptionResult struct {
 	SSES3Metadata  []byte
 }
 
-// calculatePartOffset calculates unique offset for each part to prevent IV reuse in multipart uploads
-// AWS S3 part numbers must start from 1, never 0 or negative
-func calculatePartOffset(partNumber int) int64 {
-	// AWS S3 part numbers must start from 1, never 0 or negative
-	if partNumber < 1 {
-		glog.Errorf("Invalid partNumber: %d. Must be >= 1.", partNumber)
-		return 0
-	}
-	// Using a large multiplier to ensure block offsets for different parts do not overlap.
-	// S3 part size limit is 5GB, so this provides a large safety margin.
-	partOffset := int64(partNumber-1) * s3_constants.PartOffsetMultiplier
-	return partOffset
-}
-
 // handleSSECEncryption processes SSE-C encryption for the data reader
 func (s3a *S3ApiServer) handleSSECEncryption(r *http.Request, dataReader io.Reader) (io.Reader, *SSECustomerKey, []byte, s3err.ErrorCode) {
 	// Handle SSE-C encryption if requested

weed/s3api/s3lifecycle/evaluator.go

@@ -0,0 +1,127 @@
+package s3lifecycle
+
+import "time"
+
+// Evaluate checks the given lifecycle rules against an object and returns
+// the highest-priority action that applies. The evaluation follows S3's
+// action priority:
+//  1. ExpiredObjectDeleteMarker (delete marker is sole version)
+//  2. NoncurrentVersionExpiration (non-current version age/count)
+//  3. Current version Expiration (Days or Date)
+//
+// AbortIncompleteMultipartUpload is evaluated separately since it applies
+// to uploads, not objects. Use EvaluateMPUAbort for that.
+func Evaluate(rules []Rule, obj ObjectInfo, now time.Time) EvalResult {
+	// Phase 1: ExpiredObjectDeleteMarker
+	if obj.IsDeleteMarker && obj.IsLatest && obj.NumVersions == 1 {
+		for _, rule := range rules {
+			if rule.Status != "Enabled" {
+				continue
+			}
+			if !MatchesFilter(rule, obj) {
+				continue
+			}
+			if rule.ExpiredObjectDeleteMarker {
+				return EvalResult{Action: ActionExpireDeleteMarker, RuleID: rule.ID}
+			}
+		}
+	}
+
+	// Phase 2: NoncurrentVersionExpiration
+	if !obj.IsLatest && !obj.SuccessorModTime.IsZero() {
+		for _, rule := range rules {
+			if ShouldExpireNoncurrentVersion(rule, obj, obj.NoncurrentIndex, now) {
+				return EvalResult{Action: ActionDeleteVersion, RuleID: rule.ID}
+			}
+		}
+	}
+
+	// Phase 3: Current version Expiration
+	if obj.IsLatest && !obj.IsDeleteMarker {
+		for _, rule := range rules {
+			if rule.Status != "Enabled" {
+				continue
+			}
+			if !MatchesFilter(rule, obj) {
+				continue
+			}
+			// Date-based expiration
+			if !rule.ExpirationDate.IsZero() && !now.Before(rule.ExpirationDate) {
+				return EvalResult{Action: ActionDeleteObject, RuleID: rule.ID}
+			}
+			// Days-based expiration
+			if rule.ExpirationDays > 0 {
+				expiryTime := expectedExpiryTime(obj.ModTime, rule.ExpirationDays)
+				if !now.Before(expiryTime) {
+					return EvalResult{Action: ActionDeleteObject, RuleID: rule.ID}
+				}
+			}
+		}
+	}
+
+	return EvalResult{Action: ActionNone}
+}
+
+// ShouldExpireNoncurrentVersion checks whether a non-current version should
+// be expired considering both NoncurrentDays and NewerNoncurrentVersions.
+// noncurrentIndex is the 0-based position among non-current versions sorted
+// newest-first (0 = newest non-current version).
+func ShouldExpireNoncurrentVersion(rule Rule, obj ObjectInfo, noncurrentIndex int, now time.Time) bool {
+	if rule.Status != "Enabled" {
+		return false
+	}
+	if rule.NoncurrentVersionExpirationDays <= 0 {
+		return false
+	}
+	if obj.IsLatest || obj.SuccessorModTime.IsZero() {
+		return false
+	}
+	if !MatchesFilter(rule, obj) {
+		return false
+	}
+
+	// Check age threshold.
+	expiryTime := expectedExpiryTime(obj.SuccessorModTime, rule.NoncurrentVersionExpirationDays)
+	if now.Before(expiryTime) {
+		return false
+	}
+
+	// Check NewerNoncurrentVersions count threshold.
+	if rule.NewerNoncurrentVersions > 0 && noncurrentIndex < rule.NewerNoncurrentVersions {
+		return false
+	}
+
+	return true
+}
+
+// EvaluateMPUAbort finds the applicable AbortIncompleteMultipartUpload rule
+// for a multipart upload with the given key prefix and creation time.
+func EvaluateMPUAbort(rules []Rule, uploadKey string, createdAt time.Time, now time.Time) EvalResult {
+	for _, rule := range rules {
+		if rule.Status != "Enabled" {
+			continue
+		}
+		if rule.AbortMPUDaysAfterInitiation <= 0 {
+			continue
+		}
+		if !matchesPrefix(rule.Prefix, uploadKey) {
+			continue
+		}
+		cutoff := expectedExpiryTime(createdAt, rule.AbortMPUDaysAfterInitiation)
+		if !now.Before(cutoff) {
+			return EvalResult{Action: ActionAbortMultipartUpload, RuleID: rule.ID}
+		}
+	}
+	return EvalResult{Action: ActionNone}
+}
+
+// expectedExpiryTime computes the expiration time given a reference time and
+// a number of days. Following S3 semantics, expiration happens at midnight UTC
+// of the day after the specified number of days.
+func expectedExpiryTime(refTime time.Time, days int) time.Time {
+	if days == 0 {
+		return refTime
+	}
+	t := refTime.UTC().Add(time.Duration(days+1) * 24 * time.Hour)
+	return t.Truncate(24 * time.Hour)
+}

weed/s3api/s3lifecycle/evaluator_test.go

@@ -0,0 +1,495 @@
+package s3lifecycle
+
+import (
+	"testing"
+	"time"
+)
+
+var now = time.Date(2026, 3, 27, 12, 0, 0, 0, time.UTC)
+
+func TestEvaluate_ExpirationDays(t *testing.T) {
+	rules := []Rule{{
+		ID: "expire-30d", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+
+	t.Run("object_older_than_days_is_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "data/file.txt", IsLatest: true,
+			ModTime: now.Add(-31 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+		assertEqual(t, "expire-30d", result.RuleID)
+	})
+
+	t.Run("object_younger_than_days_is_not_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "data/file.txt", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("non_latest_version_not_affected_by_expiration_days", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "data/file.txt", IsLatest: false,
+			ModTime: now.Add(-60 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("delete_marker_not_affected_by_expiration_days", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "data/file.txt", IsLatest: true, IsDeleteMarker: true,
+			ModTime: now.Add(-60 * 24 * time.Hour), NumVersions: 3,
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_ExpirationDate(t *testing.T) {
+	expirationDate := time.Date(2026, 3, 15, 0, 0, 0, 0, time.UTC)
+	rules := []Rule{{
+		ID: "expire-date", Status: "Enabled",
+		ExpirationDate: expirationDate,
+	}}
+
+	t.Run("object_expired_after_date", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-60 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+	})
+
+	t.Run("object_not_expired_before_date", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-1 * time.Hour),
+		}
+		beforeDate := time.Date(2026, 3, 10, 0, 0, 0, 0, time.UTC)
+		result := Evaluate(rules, obj, beforeDate)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_ExpiredObjectDeleteMarker(t *testing.T) {
+	rules := []Rule{{
+		ID: "cleanup-markers", Status: "Enabled",
+		ExpiredObjectDeleteMarker: true,
+	}}
+
+	t.Run("sole_delete_marker_is_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true, IsDeleteMarker: true,
+			NumVersions: 1,
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionExpireDeleteMarker, result.Action)
+	})
+
+	t.Run("delete_marker_with_other_versions_not_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true, IsDeleteMarker: true,
+			NumVersions: 3,
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("non_latest_delete_marker_not_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false, IsDeleteMarker: true,
+			NumVersions: 1,
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("non_delete_marker_not_affected", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true, IsDeleteMarker: false,
+			NumVersions: 1,
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_NoncurrentVersionExpiration(t *testing.T) {
+	rules := []Rule{{
+		ID: "expire-noncurrent", Status: "Enabled",
+		NoncurrentVersionExpirationDays: 30,
+	}}
+
+	t.Run("old_noncurrent_version_is_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false,
+			SuccessorModTime: now.Add(-45 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteVersion, result.Action)
+	})
+
+	t.Run("recent_noncurrent_version_is_not_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false,
+			SuccessorModTime: now.Add(-10 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("latest_version_not_affected", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-60 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestShouldExpireNoncurrentVersion(t *testing.T) {
+	rule := Rule{
+		ID: "noncurrent-rule", Status: "Enabled",
+		NoncurrentVersionExpirationDays: 30,
+		NewerNoncurrentVersions:         2,
+	}
+
+	t.Run("old_version_beyond_count_is_expired", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false,
+			SuccessorModTime: now.Add(-45 * 24 * time.Hour),
+		}
+		// noncurrentIndex=2 means this is the 3rd noncurrent version (0-indexed)
+		// With NewerNoncurrentVersions=2, indices 0 and 1 are kept.
+		if !ShouldExpireNoncurrentVersion(rule, obj, 2, now) {
+			t.Error("expected version at index 2 to be expired")
+		}
+	})
+
+	t.Run("old_version_within_count_is_kept", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false,
+			SuccessorModTime: now.Add(-45 * 24 * time.Hour),
+		}
+		// noncurrentIndex=1 is within the keep threshold (NewerNoncurrentVersions=2).
+		if ShouldExpireNoncurrentVersion(rule, obj, 1, now) {
+			t.Error("expected version at index 1 to be kept")
+		}
+	})
+
+	t.Run("recent_version_beyond_count_is_kept", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false,
+			SuccessorModTime: now.Add(-5 * 24 * time.Hour),
+		}
+		// Even at index 5 (beyond count), if too young, it's kept.
+		if ShouldExpireNoncurrentVersion(rule, obj, 5, now) {
+			t.Error("expected recent version to be kept regardless of index")
+		}
+	})
+
+	t.Run("disabled_rule_never_expires", func(t *testing.T) {
+		disabled := Rule{
+			ID: "disabled", Status: "Disabled",
+			NoncurrentVersionExpirationDays: 1,
+		}
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: false,
+			SuccessorModTime: now.Add(-365 * 24 * time.Hour),
+		}
+		if ShouldExpireNoncurrentVersion(disabled, obj, 10, now) {
+			t.Error("disabled rule should never expire")
+		}
+	})
+}
+
+func TestEvaluate_PrefixFilter(t *testing.T) {
+	rules := []Rule{{
+		ID: "logs-only", Status: "Enabled",
+		Prefix:         "logs/",
+		ExpirationDays: 7,
+	}}
+
+	t.Run("matching_prefix", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "logs/app.log", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+	})
+
+	t.Run("non_matching_prefix", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "data/file.txt", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_TagFilter(t *testing.T) {
+	rules := []Rule{{
+		ID: "temp-only", Status: "Enabled",
+		ExpirationDays: 1,
+		FilterTags:     map[string]string{"env": "temp"},
+	}}
+
+	t.Run("matching_tags", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-5 * 24 * time.Hour),
+			Tags:    map[string]string{"env": "temp", "project": "foo"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+	})
+
+	t.Run("missing_tag", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-5 * 24 * time.Hour),
+			Tags:    map[string]string{"project": "foo"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("wrong_tag_value", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-5 * 24 * time.Hour),
+			Tags:    map[string]string{"env": "prod"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("nil_object_tags", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true,
+			ModTime: now.Add(-5 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_SizeFilter(t *testing.T) {
+	rules := []Rule{{
+		ID: "large-files", Status: "Enabled",
+		ExpirationDays:        7,
+		FilterSizeGreaterThan: 1024 * 1024,       // > 1 MB
+		FilterSizeLessThan:    100 * 1024 * 1024,  // < 100 MB
+	}}
+
+	t.Run("matching_size", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.bin", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    10 * 1024 * 1024, // 10 MB
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+	})
+
+	t.Run("too_small", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.bin", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    512, // 512 bytes
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("too_large", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "file.bin", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    200 * 1024 * 1024, // 200 MB
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_CombinedFilters(t *testing.T) {
+	rules := []Rule{{
+		ID: "combined", Status: "Enabled",
+		Prefix:                "logs/",
+		ExpirationDays:        7,
+		FilterTags:            map[string]string{"env": "dev"},
+		FilterSizeGreaterThan: 100,
+	}}
+
+	t.Run("all_filters_match", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "logs/app.log", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    1024,
+			Tags:    map[string]string{"env": "dev"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+	})
+
+	t.Run("prefix_doesnt_match", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "data/app.log", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    1024,
+			Tags:    map[string]string{"env": "dev"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("tag_doesnt_match", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "logs/app.log", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    1024,
+			Tags:    map[string]string{"env": "prod"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("size_doesnt_match", func(t *testing.T) {
+		obj := ObjectInfo{
+			Key: "logs/app.log", IsLatest: true,
+			ModTime: now.Add(-10 * 24 * time.Hour),
+			Size:    50, // too small
+			Tags:    map[string]string{"env": "dev"},
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestEvaluate_DisabledRule(t *testing.T) {
+	rules := []Rule{{
+		ID: "disabled", Status: "Disabled",
+		ExpirationDays: 1,
+	}}
+	obj := ObjectInfo{
+		Key: "file.txt", IsLatest: true,
+		ModTime: now.Add(-365 * 24 * time.Hour),
+	}
+	result := Evaluate(rules, obj, now)
+	assertAction(t, ActionNone, result.Action)
+}
+
+func TestEvaluate_MultipleRules_Priority(t *testing.T) {
+	t.Run("delete_marker_takes_priority_over_expiration", func(t *testing.T) {
+		rules := []Rule{
+			{ID: "expire", Status: "Enabled", ExpirationDays: 1},
+			{ID: "marker", Status: "Enabled", ExpiredObjectDeleteMarker: true},
+		}
+		obj := ObjectInfo{
+			Key: "file.txt", IsLatest: true, IsDeleteMarker: true,
+			NumVersions: 1, ModTime: now.Add(-10 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionExpireDeleteMarker, result.Action)
+		assertEqual(t, "marker", result.RuleID)
+	})
+
+	t.Run("first_matching_expiration_rule_wins", func(t *testing.T) {
+		rules := []Rule{
+			{ID: "rule1", Status: "Enabled", ExpirationDays: 30, Prefix: "logs/"},
+			{ID: "rule2", Status: "Enabled", ExpirationDays: 7},
+		}
+		obj := ObjectInfo{
+			Key: "logs/app.log", IsLatest: true,
+			ModTime: now.Add(-31 * 24 * time.Hour),
+		}
+		result := Evaluate(rules, obj, now)
+		assertAction(t, ActionDeleteObject, result.Action)
+		assertEqual(t, "rule1", result.RuleID)
+	})
+}
+
+func TestEvaluate_EmptyPrefix(t *testing.T) {
+	rules := []Rule{{
+		ID: "all", Status: "Enabled",
+		ExpirationDays: 30,
+	}}
+	obj := ObjectInfo{
+		Key: "any/path/file.txt", IsLatest: true,
+		ModTime: now.Add(-31 * 24 * time.Hour),
+	}
+	result := Evaluate(rules, obj, now)
+	assertAction(t, ActionDeleteObject, result.Action)
+}
+
+func TestEvaluateMPUAbort(t *testing.T) {
+	rules := []Rule{{
+		ID: "abort-mpu", Status: "Enabled",
+		AbortMPUDaysAfterInitiation: 7,
+	}}
+
+	t.Run("old_upload_is_aborted", func(t *testing.T) {
+		result := EvaluateMPUAbort(rules, "uploads/file.bin", now.Add(-10*24*time.Hour), now)
+		assertAction(t, ActionAbortMultipartUpload, result.Action)
+	})
+
+	t.Run("recent_upload_is_not_aborted", func(t *testing.T) {
+		result := EvaluateMPUAbort(rules, "uploads/file.bin", now.Add(-3*24*time.Hour), now)
+		assertAction(t, ActionNone, result.Action)
+	})
+
+	t.Run("prefix_scoped_abort", func(t *testing.T) {
+		prefixRules := []Rule{{
+			ID: "abort-logs", Status: "Enabled",
+			Prefix:                      "logs/",
+			AbortMPUDaysAfterInitiation: 1,
+		}}
+		result := EvaluateMPUAbort(prefixRules, "data/file.bin", now.Add(-5*24*time.Hour), now)
+		assertAction(t, ActionNone, result.Action)
+	})
+}
+
+func TestExpectedExpiryTime(t *testing.T) {
+	ref := time.Date(2026, 3, 1, 15, 30, 0, 0, time.UTC)
+
+	t.Run("30_days", func(t *testing.T) {
+		// S3 spec: expires at midnight UTC of day 32 (ref + 31 days, truncated).
+		expiry := expectedExpiryTime(ref, 30)
+		expected := time.Date(2026, 4, 1, 0, 0, 0, 0, time.UTC)
+		if !expiry.Equal(expected) {
+			t.Errorf("expected %v, got %v", expected, expiry)
+		}
+	})
+
+	t.Run("zero_days_returns_ref", func(t *testing.T) {
+		expiry := expectedExpiryTime(ref, 0)
+		if !expiry.Equal(ref) {
+			t.Errorf("expected %v, got %v", ref, expiry)
+		}
+	})
+}
+
+func assertAction(t *testing.T, expected, actual Action) {
+	t.Helper()
+	if expected != actual {
+		t.Errorf("expected action %d, got %d", expected, actual)
+	}
+}
+
+func assertEqual(t *testing.T, expected, actual string) {
+	t.Helper()
+	if expected != actual {
+		t.Errorf("expected %q, got %q", expected, actual)
+	}
+}

weed/s3api/s3lifecycle/filter.go

@@ -0,0 +1,56 @@
+package s3lifecycle
+
+import "strings"
+
+// MatchesFilter checks if an object matches the rule's filter criteria
+// (prefix, tags, and size constraints).
+func MatchesFilter(rule Rule, obj ObjectInfo) bool {
+	if !matchesPrefix(rule.Prefix, obj.Key) {
+		return false
+	}
+	if !matchesTags(rule.FilterTags, obj.Tags) {
+		return false
+	}
+	if !matchesSize(rule.FilterSizeGreaterThan, rule.FilterSizeLessThan, obj.Size) {
+		return false
+	}
+	return true
+}
+
+// matchesPrefix returns true if the object key starts with the given prefix.
+// An empty prefix matches all keys.
+func matchesPrefix(prefix, key string) bool {
+	if prefix == "" {
+		return true
+	}
+	return strings.HasPrefix(key, prefix)
+}
+
+// matchesTags returns true if all rule tags are present in the object's tags
+// with matching values. An empty or nil rule tag set matches all objects.
+func matchesTags(ruleTags, objTags map[string]string) bool {
+	if len(ruleTags) == 0 {
+		return true
+	}
+	if len(objTags) == 0 {
+		return false
+	}
+	for k, v := range ruleTags {
+		if objVal, ok := objTags[k]; !ok || objVal != v {
+			return false
+		}
+	}
+	return true
+}
+
+// matchesSize returns true if the object's size falls within the specified
+// bounds. Zero values mean no constraint on that side.
+func matchesSize(greaterThan, lessThan, objSize int64) bool {
+	if greaterThan > 0 && objSize <= greaterThan {
+		return false
+	}
+	if lessThan > 0 && objSize >= lessThan {
+		return false
+	}
+	return true
+}

weed/s3api/s3lifecycle/filter_test.go

@@ -0,0 +1,79 @@
+package s3lifecycle
+
+import "testing"
+
+func TestMatchesPrefix(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+		key    string
+		want   bool
+	}{
+		{"empty_prefix_matches_all", "", "any/key.txt", true},
+		{"exact_prefix_match", "logs/", "logs/app.log", true},
+		{"prefix_mismatch", "logs/", "data/file.txt", false},
+		{"key_shorter_than_prefix", "very/long/prefix/", "short", false},
+		{"prefix_equals_key", "exact", "exact", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := matchesPrefix(tt.prefix, tt.key); got != tt.want {
+				t.Errorf("matchesPrefix(%q, %q) = %v, want %v", tt.prefix, tt.key, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestMatchesTags(t *testing.T) {
+	tests := []struct {
+		name     string
+		ruleTags map[string]string
+		objTags  map[string]string
+		want     bool
+	}{
+		{"nil_rule_tags_match_all", nil, map[string]string{"a": "1"}, true},
+		{"empty_rule_tags_match_all", map[string]string{}, map[string]string{"a": "1"}, true},
+		{"nil_obj_tags_no_match", map[string]string{"a": "1"}, nil, false},
+		{"single_tag_match", map[string]string{"env": "dev"}, map[string]string{"env": "dev", "foo": "bar"}, true},
+		{"single_tag_value_mismatch", map[string]string{"env": "dev"}, map[string]string{"env": "prod"}, false},
+		{"single_tag_key_missing", map[string]string{"env": "dev"}, map[string]string{"foo": "bar"}, false},
+		{"multi_tag_all_match", map[string]string{"env": "dev", "tier": "hot"}, map[string]string{"env": "dev", "tier": "hot", "extra": "x"}, true},
+		{"multi_tag_partial_match", map[string]string{"env": "dev", "tier": "hot"}, map[string]string{"env": "dev"}, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := matchesTags(tt.ruleTags, tt.objTags); got != tt.want {
+				t.Errorf("matchesTags() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestMatchesSize(t *testing.T) {
+	tests := []struct {
+		name        string
+		greaterThan int64
+		lessThan    int64
+		objSize     int64
+		want        bool
+	}{
+		{"no_constraints", 0, 0, 1000, true},
+		{"only_greater_than_pass", 100, 0, 200, true},
+		{"only_greater_than_fail", 100, 0, 50, false},
+		{"only_greater_than_equal_fail", 100, 0, 100, false},
+		{"only_less_than_pass", 0, 1000, 500, true},
+		{"only_less_than_fail", 0, 1000, 2000, false},
+		{"only_less_than_equal_fail", 0, 1000, 1000, false},
+		{"both_constraints_pass", 100, 1000, 500, true},
+		{"both_constraints_too_small", 100, 1000, 50, false},
+		{"both_constraints_too_large", 100, 1000, 2000, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := matchesSize(tt.greaterThan, tt.lessThan, tt.objSize); got != tt.want {
+				t.Errorf("matchesSize(%d, %d, %d) = %v, want %v",
+					tt.greaterThan, tt.lessThan, tt.objSize, got, tt.want)
+			}
+		})
+	}
+}

weed/s3api/s3lifecycle/rule.go

@@ -0,0 +1,95 @@
+package s3lifecycle
+
+import "time"
+
+// Rule is a flattened, evaluator-friendly representation of an S3 lifecycle rule.
+// Callers convert from the XML-parsed s3api.Rule (which has nested structs with
+// set-flags for conditional XML marshaling) to this type.
+type Rule struct {
+	ID     string
+	Status string // "Enabled" or "Disabled"
+
+	// Prefix filter (from Rule.Prefix or Rule.Filter.Prefix or Rule.Filter.And.Prefix).
+	Prefix string
+
+	// Expiration for current versions.
+	ExpirationDays            int
+	ExpirationDate            time.Time
+	ExpiredObjectDeleteMarker bool
+
+	// Expiration for non-current versions.
+	NoncurrentVersionExpirationDays int
+	NewerNoncurrentVersions         int
+
+	// Abort incomplete multipart uploads.
+	AbortMPUDaysAfterInitiation int
+
+	// Tag filter (from Rule.Filter.Tag or Rule.Filter.And.Tags).
+	FilterTags map[string]string
+
+	// Size filters.
+	FilterSizeGreaterThan int64
+	FilterSizeLessThan    int64
+}
+
+// ObjectInfo is the metadata about an object that the evaluator uses to
+// determine which lifecycle action applies. Callers build this from filer
+// entry attributes and extended metadata.
+type ObjectInfo struct {
+	// Key is the object key relative to the bucket root.
+	Key string
+
+	// ModTime is the object's modification time (entry.Attributes.Mtime).
+	ModTime time.Time
+
+	// Size is the object size in bytes (entry.Attributes.FileSize).
+	Size int64
+
+	// IsLatest is true if this is the current version of the object.
+	IsLatest bool
+
+	// IsDeleteMarker is true if this entry is an S3 delete marker.
+	IsDeleteMarker bool
+
+	// NumVersions is the total number of versions for this object key,
+	// including delete markers. Used for ExpiredObjectDeleteMarker evaluation.
+	NumVersions int
+
+	// SuccessorModTime is the creation time of the version that replaced
+	// this one (making it non-current). Derived from the successor's version
+	// ID timestamp. Zero value for the latest version.
+	SuccessorModTime time.Time
+
+	// NoncurrentIndex is the 0-based position among non-current versions
+	// sorted newest-first (0 = newest non-current version). Used by
+	// NewerNoncurrentVersions evaluation. -1 or unset for current versions.
+	NoncurrentIndex int
+
+	// Tags are the object's user-defined tags, extracted from the entry's
+	// Extended metadata (keys prefixed with "X-Amz-Tagging-").
+	Tags map[string]string
+}
+
+// Action represents the lifecycle action to take on an object.
+type Action int
+
+const (
+	// ActionNone means no lifecycle rule applies.
+	ActionNone Action = iota
+	// ActionDeleteObject deletes the current version of the object.
+	ActionDeleteObject
+	// ActionDeleteVersion deletes a specific non-current version.
+	ActionDeleteVersion
+	// ActionExpireDeleteMarker removes a delete marker that is the sole remaining version.
+	ActionExpireDeleteMarker
+	// ActionAbortMultipartUpload aborts an incomplete multipart upload.
+	ActionAbortMultipartUpload
+)
+
+// EvalResult is the output of lifecycle rule evaluation.
+type EvalResult struct {
+	// Action is the lifecycle action to take.
+	Action Action
+	// RuleID is the ID of the rule that triggered this action.
+	RuleID string
+}

weed/s3api/s3lifecycle/tags.go

@@ -0,0 +1,34 @@
+package s3lifecycle
+
+import "strings"
+
+const tagPrefix = "X-Amz-Tagging-"
+
+// ExtractTags extracts S3 object tags from a filer entry's Extended metadata.
+// Tags are stored with the key prefix "X-Amz-Tagging-" followed by the tag key.
+func ExtractTags(extended map[string][]byte) map[string]string {
+	if len(extended) == 0 {
+		return nil
+	}
+	var tags map[string]string
+	for k, v := range extended {
+		if strings.HasPrefix(k, tagPrefix) {
+			if tags == nil {
+				tags = make(map[string]string)
+			}
+			tags[k[len(tagPrefix):]] = string(v)
+		}
+	}
+	return tags
+}
+
+// HasTagRules returns true if any enabled rule in the set uses tag-based filtering.
+// This is used as an optimization to skip tag extraction when no rules need it.
+func HasTagRules(rules []Rule) bool {
+	for _, r := range rules {
+		if r.Status == "Enabled" && len(r.FilterTags) > 0 {
+			return true
+		}
+	}
+	return false
+}

weed/s3api/s3lifecycle/tags_test.go

@@ -0,0 +1,89 @@
+package s3lifecycle
+
+import "testing"
+
+func TestExtractTags(t *testing.T) {
+	t.Run("extracts_tags_with_prefix", func(t *testing.T) {
+		extended := map[string][]byte{
+			"X-Amz-Tagging-env":     []byte("prod"),
+			"X-Amz-Tagging-project": []byte("foo"),
+			"Content-Type":          []byte("text/plain"),
+			"X-Amz-Meta-Custom":     []byte("value"),
+		}
+		tags := ExtractTags(extended)
+		if len(tags) != 2 {
+			t.Fatalf("expected 2 tags, got %d", len(tags))
+		}
+		if tags["env"] != "prod" {
+			t.Errorf("expected env=prod, got %q", tags["env"])
+		}
+		if tags["project"] != "foo" {
+			t.Errorf("expected project=foo, got %q", tags["project"])
+		}
+	})
+
+	t.Run("nil_extended_returns_nil", func(t *testing.T) {
+		tags := ExtractTags(nil)
+		if tags != nil {
+			t.Errorf("expected nil, got %v", tags)
+		}
+	})
+
+	t.Run("no_tags_returns_nil", func(t *testing.T) {
+		extended := map[string][]byte{
+			"Content-Type": []byte("text/plain"),
+		}
+		tags := ExtractTags(extended)
+		if tags != nil {
+			t.Errorf("expected nil, got %v", tags)
+		}
+	})
+
+	t.Run("empty_tag_value", func(t *testing.T) {
+		extended := map[string][]byte{
+			"X-Amz-Tagging-empty": []byte(""),
+		}
+		tags := ExtractTags(extended)
+		if len(tags) != 1 {
+			t.Fatalf("expected 1 tag, got %d", len(tags))
+		}
+		if tags["empty"] != "" {
+			t.Errorf("expected empty value, got %q", tags["empty"])
+		}
+	})
+}
+
+func TestHasTagRules(t *testing.T) {
+	t.Run("has_tag_rules", func(t *testing.T) {
+		rules := []Rule{
+			{Status: "Enabled", FilterTags: map[string]string{"env": "dev"}},
+		}
+		if !HasTagRules(rules) {
+			t.Error("expected true")
+		}
+	})
+
+	t.Run("no_tag_rules", func(t *testing.T) {
+		rules := []Rule{
+			{Status: "Enabled", ExpirationDays: 30},
+		}
+		if HasTagRules(rules) {
+			t.Error("expected false")
+		}
+	})
+
+	t.Run("disabled_tag_rule", func(t *testing.T) {
+		rules := []Rule{
+			{Status: "Disabled", FilterTags: map[string]string{"env": "dev"}},
+		}
+		if HasTagRules(rules) {
+			t.Error("expected false for disabled rule")
+		}
+	})
+
+	t.Run("empty_rules", func(t *testing.T) {
+		if HasTagRules(nil) {
+			t.Error("expected false for nil rules")
+		}
+	})
+}

weed/s3api/s3lifecycle/version_time.go

@@ -0,0 +1,99 @@
+package s3lifecycle
+
+import (
+	"math"
+	"strconv"
+	"time"
+)
+
+// versionIdFormatThreshold distinguishes old vs new format version IDs.
+// New format (inverted timestamps) produces values above this threshold;
+// old format (raw timestamps) produces values below it.
+const versionIdFormatThreshold = 0x4000000000000000
+
+// GetVersionTimestamp extracts the actual timestamp from a SeaweedFS version ID,
+// handling both old (raw nanosecond) and new (inverted nanosecond) formats.
+// Returns zero time if the version ID is invalid or "null".
+func GetVersionTimestamp(versionId string) time.Time {
+	ns := getVersionTimestampNanos(versionId)
+	if ns == 0 {
+		return time.Time{}
+	}
+	return time.Unix(0, ns)
+}
+
+// getVersionTimestampNanos extracts the raw nanosecond timestamp from a version ID.
+func getVersionTimestampNanos(versionId string) int64 {
+	if len(versionId) < 16 || versionId == "null" {
+		return 0
+	}
+	timestampPart, err := strconv.ParseUint(versionId[:16], 16, 64)
+	if err != nil {
+		return 0
+	}
+	if timestampPart > math.MaxInt64 {
+		return 0
+	}
+	if timestampPart > versionIdFormatThreshold {
+		// New format: inverted timestamp, convert back.
+		return int64(math.MaxInt64 - timestampPart)
+	}
+	return int64(timestampPart)
+}
+
+// isNewFormatVersionId returns true if the version ID uses inverted timestamps.
+func isNewFormatVersionId(versionId string) bool {
+	if len(versionId) < 16 || versionId == "null" {
+		return false
+	}
+	timestampPart, err := strconv.ParseUint(versionId[:16], 16, 64)
+	if err != nil {
+		return false
+	}
+	return timestampPart > versionIdFormatThreshold && timestampPart <= math.MaxInt64
+}
+
+// CompareVersionIds compares two version IDs for sorting (newest first).
+// Returns negative if a is newer, positive if b is newer, 0 if equal.
+// Handles both old and new format version IDs and uses full lexicographic
+// comparison (not just timestamps) to break ties from the random suffix.
+func CompareVersionIds(a, b string) int {
+	if a == b {
+		return 0
+	}
+	if a == "null" {
+		return 1
+	}
+	if b == "null" {
+		return -1
+	}
+
+	aIsNew := isNewFormatVersionId(a)
+	bIsNew := isNewFormatVersionId(b)
+
+	if aIsNew == bIsNew {
+		if aIsNew {
+			// New format: smaller hex = newer (inverted timestamps).
+			if a < b {
+				return -1
+			}
+			return 1
+		}
+		// Old format: smaller hex = older.
+		if a < b {
+			return 1
+		}
+		return -1
+	}
+
+	// Mixed formats: compare by actual timestamp.
+	aTime := getVersionTimestampNanos(a)
+	bTime := getVersionTimestampNanos(b)
+	if aTime > bTime {
+		return -1
+	}
+	if aTime < bTime {
+		return 1
+	}
+	return 0
+}

weed/s3api/s3lifecycle/version_time_test.go

@@ -0,0 +1,74 @@
+package s3lifecycle
+
+import (
+	"fmt"
+	"math"
+	"testing"
+	"time"
+)
+
+func TestGetVersionTimestamp(t *testing.T) {
+	t.Run("new_format_inverted_timestamp", func(t *testing.T) {
+		// Simulate a new-format version ID (inverted timestamp above threshold).
+		now := time.Now()
+		inverted := math.MaxInt64 - now.UnixNano()
+		versionId := fmt.Sprintf("%016x", inverted) + "0000000000000000"
+
+		got := GetVersionTimestamp(versionId)
+		// Should recover the original timestamp within 1 second.
+		diff := got.Sub(now)
+		if diff < -time.Second || diff > time.Second {
+			t.Errorf("timestamp diff too large: %v (got %v, want ~%v)", diff, got, now)
+		}
+	})
+
+	t.Run("old_format_raw_timestamp", func(t *testing.T) {
+		// Simulate an old-format version ID (raw nanosecond timestamp below threshold).
+		// Use a timestamp from 2023 which would be below threshold.
+		ts := time.Date(2023, 6, 15, 12, 0, 0, 0, time.UTC)
+		versionId := fmt.Sprintf("%016x", ts.UnixNano()) + "abcdef0123456789"
+
+		got := GetVersionTimestamp(versionId)
+		if !got.Equal(ts) {
+			t.Errorf("expected %v, got %v", ts, got)
+		}
+	})
+
+	t.Run("null_version_id", func(t *testing.T) {
+		got := GetVersionTimestamp("null")
+		if !got.IsZero() {
+			t.Errorf("expected zero time for null version, got %v", got)
+		}
+	})
+
+	t.Run("empty_version_id", func(t *testing.T) {
+		got := GetVersionTimestamp("")
+		if !got.IsZero() {
+			t.Errorf("expected zero time for empty version, got %v", got)
+		}
+	})
+
+	t.Run("short_version_id", func(t *testing.T) {
+		got := GetVersionTimestamp("abc")
+		if !got.IsZero() {
+			t.Errorf("expected zero time for short version, got %v", got)
+		}
+	})
+
+	t.Run("high_bit_overflow_returns_zero", func(t *testing.T) {
+		// Version ID with first 16 hex chars > math.MaxInt64 should return zero,
+		// not a wrapped negative timestamp.
+		versionId := "80000000000000000000000000000000"
+		got := GetVersionTimestamp(versionId)
+		if !got.IsZero() {
+			t.Errorf("expected zero time for overflow version ID, got %v", got)
+		}
+	})
+
+	t.Run("invalid_hex", func(t *testing.T) {
+		got := GetVersionTimestamp("zzzzzzzzzzzzzzzz0000000000000000")
+		if !got.IsZero() {
+			t.Errorf("expected zero time for invalid hex, got %v", got)
+		}
+	})
+}

weed/worker/tasks/balance/detection.go

@@ -363,6 +363,23 @@ func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetric
 		return nil, ""
 	}
 
+	// Parse replica placement once for use in both scoring and validation.
+	var volumeRP *super_block.ReplicaPlacement
+	if selectedVolume.ExpectedReplicas > 0 && selectedVolume.ExpectedReplicas <= 255 {
+		if parsed, rpErr := super_block.NewReplicaPlacementFromByte(byte(selectedVolume.ExpectedReplicas)); rpErr == nil && parsed.HasReplication() {
+			volumeRP = parsed
+		}
+	}
+
+	var replicas []types.ReplicaLocation
+	if clusterInfo.VolumeReplicaMap != nil {
+		replicas = clusterInfo.VolumeReplicaMap[selectedVolume.VolumeID]
+		if volumeRP != nil && len(replicas) == 0 {
+			glog.V(1).Infof("BALANCE [%s]: No replica locations found for volume %d, skipping placement validation",
+				diskType, selectedVolume.VolumeID)
+		}
+	}
+
 	// Resolve the target server chosen by the detection loop's effective counts.
 	// This keeps destination selection in sync with the greedy algorithm rather
 	// than relying on topology LoadCount which can diverge across iterations.
@@ -371,62 +388,43 @@ func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetric
 		// Fall back to score-based planning if the preferred target can't be resolved
 		glog.V(1).Infof("BALANCE [%s]: Cannot resolve target %s for volume %d, falling back to score-based planning: %v",
 			diskType, targetServer, selectedVolume.VolumeID, err)
-		destinationPlan, err = planBalanceDestination(clusterInfo.ActiveTopology, selectedVolume)
+		destinationPlan, err = planBalanceDestination(clusterInfo.ActiveTopology, selectedVolume, volumeRP, replicas, allowedServers)
 		if err != nil {
 			glog.Warningf("Failed to plan balance destination for volume %d: %v", selectedVolume.VolumeID, err)
 			return nil, ""
 		}
 	}
 
-	// Verify the destination is within the filtered scope. When DC/rack/node
-	// filters are active, allowedServers contains only the servers that passed
-	// filtering. The fallback planner queries the full topology, so this check
-	// prevents out-of-scope targets from leaking through.
-	if len(allowedServers) > 0 {
-		if _, ok := allowedServers[destinationPlan.TargetNode]; !ok {
-			glog.V(1).Infof("BALANCE [%s]: Planned destination %s for volume %d is outside filtered scope, skipping",
+	// Verify the resolved destination. If it falls outside the filtered scope
+	// or violates replica placement, fall back to the score-based planner and
+	// pick the best candidate that is actually valid.
+	if !isValidBalanceDestination(destinationPlan, allowedServers, volumeRP, replicas, selectedVolume.Server) {
+		switch {
+		case destinationPlan == nil:
+			glog.V(1).Infof("BALANCE [%s]: Planned destination for volume %d is nil, falling back",
+				diskType, selectedVolume.VolumeID)
+		case !isAllowedBalanceTarget(destinationPlan.TargetNode, allowedServers):
+			glog.V(1).Infof("BALANCE [%s]: Planned destination %s for volume %d is outside filtered scope, falling back",
 				diskType, destinationPlan.TargetNode, selectedVolume.VolumeID)
+		case volumeRP != nil && len(replicas) > 0:
+			glog.V(1).Infof("BALANCE [%s]: Destination %s violates replica placement for volume %d (rp=%03d), falling back",
+				diskType, destinationPlan.TargetNode, selectedVolume.VolumeID, selectedVolume.ExpectedReplicas)
+		}
+
+		destinationPlan, err = planBalanceDestination(clusterInfo.ActiveTopology, selectedVolume, volumeRP, replicas, allowedServers)
+		if err != nil {
+			glog.Warningf("BALANCE [%s]: Failed to plan fallback destination for volume %d: %v", diskType, selectedVolume.VolumeID, err)
 			return nil, ""
 		}
-	}
-
-	// Validate move against replica placement policy
-	if selectedVolume.ExpectedReplicas > 0 && selectedVolume.ExpectedReplicas <= 255 && clusterInfo.VolumeReplicaMap != nil {
-		rpBytes, rpErr := super_block.NewReplicaPlacementFromByte(byte(selectedVolume.ExpectedReplicas))
-		if rpErr == nil && rpBytes.HasReplication() {
-			replicas := clusterInfo.VolumeReplicaMap[selectedVolume.VolumeID]
-			if len(replicas) == 0 {
-				glog.V(1).Infof("BALANCE [%s]: No replica locations found for volume %d, skipping placement validation",
+		if !isValidBalanceDestination(destinationPlan, allowedServers, volumeRP, replicas, selectedVolume.Server) {
+			if destinationPlan == nil {
+				glog.V(1).Infof("BALANCE [%s]: Fallback destination for volume %d is nil",
 					diskType, selectedVolume.VolumeID)
 			} else {
-				validateMove := func(plan *topology.DestinationPlan) bool {
-					if plan == nil {
-						return false
-					}
-					target := types.ReplicaLocation{
-						DataCenter: plan.TargetDC,
-						Rack:       plan.TargetRack,
-						NodeID:     plan.TargetNode,
-					}
-					return IsGoodMove(rpBytes, replicas, selectedVolume.Server, target)
-				}
-
-				if !validateMove(destinationPlan) {
-					glog.V(1).Infof("BALANCE [%s]: Destination %s violates replica placement for volume %d (rp=%03d), falling back",
-						diskType, destinationPlan.TargetNode, selectedVolume.VolumeID, selectedVolume.ExpectedReplicas)
-					// Fall back to score-based planning
-					destinationPlan, err = planBalanceDestination(clusterInfo.ActiveTopology, selectedVolume)
-					if err != nil {
-						glog.Warningf("BALANCE [%s]: Failed to plan fallback destination for volume %d: %v", diskType, selectedVolume.VolumeID, err)
-						return nil, ""
-					}
-					if !validateMove(destinationPlan) {
-						glog.V(1).Infof("BALANCE [%s]: Fallback destination %s also violates replica placement for volume %d",
-							diskType, destinationPlan.TargetNode, selectedVolume.VolumeID)
-						return nil, ""
-					}
-				}
+				glog.V(1).Infof("BALANCE [%s]: Fallback destination %s is not valid for volume %d",
+					diskType, destinationPlan.TargetNode, selectedVolume.VolumeID)
 			}
+			return nil, ""
 		}
 	}
 
@@ -555,7 +553,10 @@ func resolveBalanceDestination(activeTopology *topology.ActiveTopology, selected
 // planBalanceDestination plans the destination for a balance operation using
 // score-based selection. Used as a fallback when the preferred target cannot
 // be resolved, and for single-move scenarios outside the detection loop.
-func planBalanceDestination(activeTopology *topology.ActiveTopology, selectedVolume *types.VolumeHealthMetrics) (*topology.DestinationPlan, error) {
+// rp may be nil when the volume has no replication constraint. When replica
+// locations are known, candidates that would violate placement are filtered
+// out before scoring.
+func planBalanceDestination(activeTopology *topology.ActiveTopology, selectedVolume *types.VolumeHealthMetrics, rp *super_block.ReplicaPlacement, replicas []types.ReplicaLocation, allowedServers map[string]int) (*topology.DestinationPlan, error) {
 	// Get source node information from topology
 	var sourceRack, sourceDC string
 
@@ -604,8 +605,21 @@ func planBalanceDestination(activeTopology *topology.ActiveTopology, selectedVol
 		if disk.DiskType != selectedVolume.DiskType {
 			continue
 		}
+		if !isAllowedBalanceTarget(disk.NodeID, allowedServers) {
+			continue
+		}
+		if rp != nil && len(replicas) > 0 {
+			target := types.ReplicaLocation{
+				DataCenter: disk.DataCenter,
+				Rack:       disk.Rack,
+				NodeID:     disk.NodeID,
+			}
+			if !IsGoodMove(rp, replicas, selectedVolume.Server, target) {
+				continue
+			}
+		}
 
-		score := calculateBalanceScore(disk, sourceRack, sourceDC, selectedVolume.Size)
+		score := calculateBalanceScore(disk, sourceRack, sourceDC, selectedVolume.Size, rp)
 		if score > bestScore {
 			bestScore = score
 			bestDisk = disk
@@ -636,7 +650,9 @@ func planBalanceDestination(activeTopology *topology.ActiveTopology, selectedVol
 // calculateBalanceScore calculates placement score for balance operations.
 // LoadCount reflects pending+assigned tasks on the disk, so we factor it into
 // the utilization estimate to avoid stacking multiple moves onto the same target.
-func calculateBalanceScore(disk *topology.DiskInfo, sourceRack, sourceDC string, volumeSize uint64) float64 {
+// rp may be nil when the volume has no replication constraint; in that case the
+// scorer defaults to preferring cross-rack/DC distribution.
+func calculateBalanceScore(disk *topology.DiskInfo, sourceRack, sourceDC string, volumeSize uint64, rp *super_block.ReplicaPlacement) float64 {
 	if disk.DiskInfo == nil {
 		return 0.0
 	}
@@ -652,17 +668,62 @@ func calculateBalanceScore(disk *topology.DiskInfo, sourceRack, sourceDC string,
 		score += (1.0 - utilization) * 50.0 // Up to 50 points for low utilization
 	}
 
-	// Prefer different racks for better distribution
-	if disk.Rack != sourceRack {
-		score += 30.0
+	// Rack scoring: respect the replication policy.
+	// If replicas must stay on the same rack (SameRackCount > 0 with no
+	// cross-rack requirement), prefer same-rack destinations. Otherwise
+	// prefer different racks for better distribution.
+	sameRack := disk.Rack == sourceRack
+	if rp != nil && rp.DiffRackCount == 0 && rp.SameRackCount > 0 {
+		if sameRack {
+			score += 30.0
+		}
+	} else {
+		if !sameRack {
+			score += 30.0
+		}
 	}
 
-	// Prefer different data centers for better distribution
-	if disk.DataCenter != sourceDC {
-		score += 20.0
+	// DC scoring: same idea. If the policy requires all copies in one DC,
+	// prefer same-DC destinations. Otherwise prefer different DCs.
+	sameDC := disk.DataCenter == sourceDC
+	if rp != nil && rp.DiffDataCenterCount == 0 && (rp.SameRackCount > 0 || rp.DiffRackCount > 0) {
+		if sameDC {
+			score += 20.0
+		}
+	} else {
+		if !sameDC {
+			score += 20.0
+		}
 	}
 
 	return score
 }
 
+func isAllowedBalanceTarget(nodeID string, allowedServers map[string]int) bool {
+	if len(allowedServers) == 0 {
+		return true
+	}
+	_, ok := allowedServers[nodeID]
+	return ok
+}
+
+func isValidBalanceDestination(plan *topology.DestinationPlan, allowedServers map[string]int, rp *super_block.ReplicaPlacement, replicas []types.ReplicaLocation, sourceNodeID string) bool {
+	if plan == nil {
+		return false
+	}
+	if !isAllowedBalanceTarget(plan.TargetNode, allowedServers) {
+		return false
+	}
+	if rp == nil || len(replicas) == 0 {
+		return true
+	}
+
+	target := types.ReplicaLocation{
+		DataCenter: plan.TargetDC,
+		Rack:       plan.TargetRack,
+		NodeID:     plan.TargetNode,
+	}
+	return IsGoodMove(rp, replicas, sourceNodeID, target)
+}
+
 // parseCSVSet splits a comma-separated string into a set of trimmed, non-empty values.

weed/worker/tasks/balance/replica_placement_test.go

@@ -3,6 +3,8 @@ package balance
 import (
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types"
 )
@@ -125,3 +127,112 @@ func TestIsGoodMove_NilReplicaPlacement(t *testing.T) {
 		t.Error("nil replica placement should allow any move")
 	}
 }
+
+func TestCalculateBalanceScore_ReplicationAware(t *testing.T) {
+	disk := func(dc, rack string) *topology.DiskInfo {
+		return &topology.DiskInfo{
+			DataCenter: dc,
+			Rack:       rack,
+			DiskInfo:   &master_pb.DiskInfo{MaxVolumeCount: 100, VolumeCount: 50},
+		}
+	}
+
+	// 001: same-rack replication — should prefer same rack and same DC
+	rp001 := rp(t, "001")
+	sameRack := calculateBalanceScore(disk("dc1", "r1"), "r1", "dc1", 0, rp001)
+	diffRack := calculateBalanceScore(disk("dc1", "r2"), "r1", "dc1", 0, rp001)
+	diffDC := calculateBalanceScore(disk("dc2", "r2"), "r1", "dc1", 0, rp001)
+	if sameRack <= diffRack {
+		t.Errorf("001: same-rack score (%v) should exceed different-rack score (%v)", sameRack, diffRack)
+	}
+	if sameRack <= diffDC {
+		t.Errorf("001: same-rack score (%v) should exceed different-DC score (%v)", sameRack, diffDC)
+	}
+
+	// 010: different-rack replication — should prefer different rack, same DC
+	rp010 := rp(t, "010")
+	sameRack = calculateBalanceScore(disk("dc1", "r1"), "r1", "dc1", 0, rp010)
+	diffRack = calculateBalanceScore(disk("dc1", "r2"), "r1", "dc1", 0, rp010)
+	if diffRack <= sameRack {
+		t.Errorf("010: different-rack score (%v) should exceed same-rack score (%v)", diffRack, sameRack)
+	}
+
+	// 100: different-DC replication — should prefer different DC
+	rp100 := rp(t, "100")
+	sameDC := calculateBalanceScore(disk("dc1", "r2"), "r1", "dc1", 0, rp100)
+	diffDCScore := calculateBalanceScore(disk("dc2", "r2"), "r1", "dc1", 0, rp100)
+	if diffDCScore <= sameDC {
+		t.Errorf("100: different-DC score (%v) should exceed same-DC score (%v)", diffDCScore, sameDC)
+	}
+
+	// nil rp: should prefer cross-rack/DC (default behavior)
+	sameRack = calculateBalanceScore(disk("dc1", "r1"), "r1", "dc1", 0, nil)
+	diffRack = calculateBalanceScore(disk("dc1", "r2"), "r1", "dc1", 0, nil)
+	if diffRack <= sameRack {
+		t.Errorf("nil rp: different-rack score (%v) should exceed same-rack score (%v)", diffRack, sameRack)
+	}
+}
+
+func TestPlanBalanceDestination_ChoosesBestValidCompositeDestination(t *testing.T) {
+	servers := []serverSpec{
+		{id: "node-a", diskType: "hdd", diskID: 1, dc: "dc1", rack: "rack1"},
+		{id: "node-b", diskType: "hdd", diskID: 2, dc: "dc1", rack: "rack3"},
+		{id: "node-c", diskType: "hdd", diskID: 3, dc: "dc1", rack: "rack1"},
+	}
+	volumes := makeVolumesWith("node-a", "hdd", "dc1", "rack1", "c1", 1, 1, withReplicas(11))
+	replicas := []types.ReplicaLocation{
+		loc("dc1", "rack1", "node-a"),
+		loc("dc1", "rack1", "node-d"),
+		loc("dc1", "rack2", "node-e"),
+	}
+
+	plan, err := planBalanceDestination(buildTopology(servers, volumes), volumes[0], rp(t, "011"), replicas, map[string]int{
+		"node-b": 0,
+		"node-c": 0,
+	})
+	if err != nil {
+		t.Fatalf("planBalanceDestination failed: %v", err)
+	}
+	if plan.TargetNode != "node-c" {
+		t.Fatalf("expected valid same-rack destination node-c, got %s", plan.TargetNode)
+	}
+	if plan.TargetRack != "rack1" {
+		t.Fatalf("expected rack1 destination, got %s", plan.TargetRack)
+	}
+}
+
+func TestCreateBalanceTask_FallbackSelectsValidCompositeDestination(t *testing.T) {
+	servers := []serverSpec{
+		{id: "node-a", diskType: "hdd", diskID: 1, dc: "dc1", rack: "rack1"},
+		{id: "node-b", diskType: "hdd", diskID: 2, dc: "dc1", rack: "rack3"},
+		{id: "node-c", diskType: "hdd", diskID: 3, dc: "dc1", rack: "rack1"},
+	}
+	volumes := makeVolumesWith("node-a", "hdd", "dc1", "rack1", "c1", 1, 1, withReplicas(11))
+	clusterInfo := &types.ClusterInfo{
+		ActiveTopology: buildTopology(servers, volumes),
+		VolumeReplicaMap: map[uint32][]types.ReplicaLocation{
+			1: {
+				loc("dc1", "rack1", "node-a"),
+				loc("dc1", "rack1", "node-d"),
+				loc("dc1", "rack2", "node-e"),
+			},
+		},
+	}
+
+	task, destination := createBalanceTask("hdd", volumes[0], clusterInfo, "node-b", map[string]int{
+		"node-b": 0,
+		"node-c": 0,
+	})
+	if task == nil {
+		t.Fatal("expected a balance task")
+	}
+	if destination != "node-c" {
+		t.Fatalf("expected fallback destination node-c, got %s", destination)
+	}
+	if len(task.TypedParams.Targets) != 1 {
+		t.Fatalf("expected 1 target, got %d", len(task.TypedParams.Targets))
+	}
+	if got := task.TypedParams.Targets[0].Node; got != "node-c:8080" {
+		t.Fatalf("expected target node-c:8080, got %s", got)
+	}
+}