ci(s3/lifecycle): wire Layer 3 integration test into PR CI

The Layer 3 integration test at test/s3/lifecycle (event-driven
lifecycle worker driven by s3.lifecycle.run-shard) only ran via local
make test-with-server. Add a dedicated workflow with a path filter
scoped to lifecycle source so PRs touching the worker, the bucket
lifecycle config handler, the lifecycle XML codec, the worker task,
or the run-shard shell command actually exercise it.

.github/workflows/s3-lifecycle-tests.yml

@@ -0,0 +1,80 @@
+name: "S3 Lifecycle Tests"
+
+on:
+  pull_request:
+    paths:
+      - 'weed/s3api/s3lifecycle/**'
+      - 'weed/s3api/s3api_internal_lifecycle.go'
+      - 'weed/s3api/s3api_internal_lifecycle_test.go'
+      - 'weed/s3api/s3api_object_lifecycle_ttl*.go'
+      - 'weed/s3api/s3api_bucket_handlers.go'
+      - 'weed/s3api/s3api_bucket_lifecycle_config.go'
+      - 'weed/s3api/lifecycle_xml/**'
+      - 'weed/worker/tasks/s3_lifecycle/**'
+      - 'weed/shell/command_s3_lifecycle_*.go'
+      - 'test/s3/lifecycle/**'
+      - '.github/workflows/s3-lifecycle-tests.yml'
+  push:
+    branches: [ master, main ]
+    paths:
+      - 'weed/s3api/s3lifecycle/**'
+      - 'weed/s3api/s3api_internal_lifecycle.go'
+      - 'weed/s3api/s3api_internal_lifecycle_test.go'
+      - 'weed/s3api/s3api_object_lifecycle_ttl*.go'
+      - 'weed/s3api/s3api_bucket_handlers.go'
+      - 'weed/s3api/s3api_bucket_lifecycle_config.go'
+      - 'weed/s3api/lifecycle_xml/**'
+      - 'weed/worker/tasks/s3_lifecycle/**'
+      - 'weed/shell/command_s3_lifecycle_*.go'
+      - 'test/s3/lifecycle/**'
+      - '.github/workflows/s3-lifecycle-tests.yml'
+
+concurrency:
+  group: ${{ github.head_ref }}/s3-lifecycle-tests
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  s3-lifecycle-integration:
+    name: S3 Lifecycle Integration 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 Lifecycle Tests
+        timeout-minutes: 15
+        working-directory: test/s3/lifecycle
+        run: |
+          set -x
+          make test-with-server
+
+      - name: Show server logs on failure
+        if: failure()
+        working-directory: test/s3/lifecycle
+        run: |
+          echo "=== Server Logs ==="
+          if [ -f weed-test.log ]; then
+            tail -200 weed-test.log
+          else
+            echo "No server log file found"
+          fi
+          ps aux | grep -E "(weed|test)" || true
+          netstat -tlnp 2>/dev/null | grep -E "(8333|9333|8080|8888)" || true
+
+      - name: Upload test logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: s3-lifecycle-test-logs
+          path: test/s3/lifecycle/weed-test*.log
+          retention-days: 3

weed/s3api/s3api_internal_lifecycle.go

@@ -120,6 +120,15 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 				return noopResolved("VERSION_IS_LATEST"), nil
 			}
 		}
+		// Re-check sole-survivor: a fresh PUT can land between schedule
+		// and dispatch. Identity-CAS upstream covers the marker bytes;
+		// this covers the directory shape.
+		if req.ActionKind == s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER {
+			outcome, err := s3a.checkSoleSurvivorMarker(ctx, req.Bucket, req.ObjectPath, req.VersionId)
+			if outcome != nil || err != nil {
+				return outcome, err
+			}
+		}
 		if err := s3a.deleteSpecificObjectVersion(req.Bucket, req.ObjectPath, req.VersionId); err != nil {
 			if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrObjectNotFound) {
 				return noopResolved("NOT_FOUND_AT_DELETE"), nil
@@ -175,6 +184,92 @@ func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle
 	return done(), nil
 }
 
+// checkSoleSurvivorMarker returns nil to proceed with the delete, or a
+// terminal response when state has drifted: count != 1, the surviving
+// entry is a different version, the .versions/ directory's latest
+// pointer doesn't name versionId, or a bare null-version exists outside
+// .versions/. Pointer missing while a marker is present is treated as
+// retry-later — the create races with the directory metadata update.
+func (s3a *S3ApiServer) checkSoleSurvivorMarker(ctx context.Context, bucket, object, versionId string) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+	bucketDir := s3a.bucketDir(bucket)
+	versionsDir := bucketDir + "/" + object + s3_constants.VersionsFolder
+	count := 0
+	var firstName string
+	err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return filer_pb.SeaweedList(ctx, client, versionsDir, "", func(entry *filer_pb.Entry, _ bool) error {
+			count++
+			if count == 1 && entry != nil {
+				firstName = entry.Name
+			}
+			return nil
+		}, "", false, 2)
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return noopResolved("NOT_FOUND"), nil
+		}
+		return retryLater("TRANSPORT_ERROR: sole-survivor list: " + err.Error()), nil
+	}
+	if count == 0 {
+		return noopResolved("NOT_FOUND"), nil
+	}
+	if count > 1 {
+		return noopResolved("NOT_SOLE_SURVIVOR"), nil
+	}
+	// SeaweedList delivered a single callback but with a nil entry; we
+	// can't compare names so retry rather than silently bypass the
+	// marker-replaced check.
+	if firstName == "" {
+		return retryLater("PENDING_SURVIVOR_ENTRY"), nil
+	}
+	if versionId != "" && firstName != s3a.getVersionFileName(versionId) {
+		return noopResolved("MARKER_REPLACED"), nil
+	}
+	// Latest-pointer check: createDeleteMarker writes the marker file
+	// and then updates the parent directory's Extended map. Reading
+	// before the second step lands would see count==1 but no pointer;
+	// retry-later rather than mistakenly delete.
+	parent, name := path.Split(versionsDir)
+	parent = strings.TrimRight(parent, "/")
+	if parent == "" {
+		parent = "/"
+	}
+	versionsEntry, err := s3a.getEntry(parent, name)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return noopResolved("NOT_FOUND"), nil
+		}
+		return retryLater("TRANSPORT_ERROR: latest-pointer lookup: " + err.Error()), nil
+	}
+	if versionsEntry == nil {
+		return noopResolved("NOT_FOUND"), nil
+	}
+	latest, hasPointer := versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey]
+	if !hasPointer || len(latest) == 0 {
+		return retryLater("PENDING_LATEST_POINTER"), nil
+	}
+	if string(latest) != versionId {
+		return noopResolved("MARKER_NOT_LATEST"), nil
+	}
+	// Null-version check: pre-versioning objects survive as the bare
+	// <bucket>/<key>. Both regular files and explicit S3 directory-key
+	// markers (object names ending in /) qualify; the listing path
+	// (s3api_object_versioning.go processExplicitDirectory) treats both
+	// as the null version. getEntry uses NewFullPath so a trailing slash
+	// in object splits the same as a regular key.
+	bareEntry, err := s3a.getEntry(bucketDir, object)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return nil, nil
+		}
+		return retryLater("TRANSPORT_ERROR: null-version lookup: " + err.Error()), nil
+	}
+	if bareEntry != nil && (!bareEntry.IsDirectory || bareEntry.IsDirectoryKeyObject()) {
+		return noopResolved("NULL_VERSION_PRESENT"), nil
+	}
+	return nil, nil
+}
+
 // isCurrentLatestVersion reports whether versionId is the version the
 // .versions/ directory currently points to. SeaweedFS records the latest
 // version on the parent directory's Extended map; without consulting it,

weed/s3api/s3api_object_handlers_put.go

@@ -1341,9 +1341,14 @@ func (s3a *S3ApiServer) updateIsLatestFlagsForSuspendedVersioning(bucket, object
 	// Clear the latest version metadata from .versions directory since "null" is now latest
 	versionsEntry, err := s3a.getEntry(bucketDir, versionsObjectPath)
 	if err == nil && versionsEntry.Extended != nil {
-		// Remove latest version metadata so all versions show IsLatest=false
+		// Remove latest version metadata so all versions show IsLatest=false.
+		// Also wipe cached list-metadata (size/mtime/etag/owner/delete-marker):
+		// they were stamped from the prior latest, and stale cached mtime
+		// would let lifecycle compute SuccessorModTime off the displaced
+		// version's age rather than today's null PUT.
 		delete(versionsEntry.Extended, s3_constants.ExtLatestVersionIdKey)
 		delete(versionsEntry.Extended, s3_constants.ExtLatestVersionFileNameKey)
+		clearCachedVersionMetadata(versionsEntry.Extended)
 
 		// Update the .versions directory entry
 		err = s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) {

weed/s3api/s3lifecycle/dispatcher/pipeline.go

@@ -205,6 +205,8 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		cancel() // wake the dispatcher goroutine to drain & exit
 	}()
 
+	lister := &filerSiblingLister{client: p.FilerClient, bucketsPath: p.BucketsPath}
+
 	// Router/dispatcher goroutine: pulls events, routes them to per-shard
 	// schedules, ticks every shard's dispatcher on the same cadence, and
 	// checkpoints every shard's cursor on the checkpoint cadence. One
@@ -257,7 +259,7 @@ func (p *Pipeline) Run(ctx context.Context) error {
 				// against the prior (empty) snapshot would silently drop
 				// every match. Engine.Snapshot is an atomic Load.
 				snap := p.Engine.Snapshot()
-				for _, m := range router.Route(snap, ev, time.Now()) {
+				for _, m := range router.Route(runCtx, snap, ev, time.Now(), lister) {
 					st.dispatch.Schedule.Add(m)
 				}
 			case <-dt.C:

weed/s3api/s3lifecycle/dispatcher/pipeline_test.go

@@ -48,7 +48,7 @@ func TestPipelineIntegrationInMemory(t *testing.T) {
 		},
 	}
 
-	matches := router.Route(snap, ev, now)
+	matches := router.Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %v", matches)
 	}

weed/s3api/s3lifecycle/dispatcher/sibling_lister.go

@@ -0,0 +1,162 @@
+package dispatcher
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// filerSiblingLister inspects the .versions/<key>/ folder and the bare
+// null-version. Listing caps at 2 — callers only distinguish 1 vs >=2.
+type filerSiblingLister struct {
+	client      filer_pb.SeaweedFilerClient
+	bucketsPath string
+}
+
+func (l *filerSiblingLister) Survivors(ctx context.Context, bucket, objectKey string) (router.Survivors, error) {
+	bucketPath := strings.TrimSuffix(l.bucketsPath, "/") + "/" + bucket
+	versionsDir := bucketPath + "/" + objectKey + s3_constants.VersionsFolder
+
+	var s router.Survivors
+	err := filer_pb.SeaweedList(ctx, l.client, versionsDir, "", func(entry *filer_pb.Entry, _ bool) error {
+		s.Count++
+		if s.Count == 1 && entry != nil {
+			s.LoneEntry = entry
+		} else if s.Count > 1 {
+			s.LoneEntry = nil
+		}
+		return nil
+	}, "", false, 2)
+	if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
+		return router.Survivors{}, err
+	}
+
+	// NewFullPath strips a trailing slash from objectKey so directory-key
+	// objects (foo/) split the same as regular keys. LookupEntry
+	// normalizes the gRPC string-mapped not-found into ErrNotFound; a
+	// raw client.LookupDirectoryEntry would return that as a generic
+	// error and suppress every otherwise-valid match.
+	parent, name := util.NewFullPath(bucketPath, objectKey).DirAndName()
+	resp, err := filer_pb.LookupEntry(ctx, l.client, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: parent,
+		Name:      name,
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return s, nil
+		}
+		return router.Survivors{}, err
+	}
+	// Bare regular file or an explicit S3 directory-marker (an empty
+	// directory entry with Mime set) both count as the null version.
+	if resp.Entry != nil && (!resp.Entry.IsDirectory || resp.Entry.IsDirectoryKeyObject()) {
+		s.HasNullVersion = true
+	}
+	return s, nil
+}
+
+// ListVersions paginates every file under .versions/<key>/. Used by
+// the pointer-transition expansion path when a NewerNoncurrentVersions
+// rule needs accurate per-version ranking. Subdirectories and entries
+// without ExtVersionIdKey are filtered out. NotFound returns
+// (nil, nil) so a hard-deleted .versions/ container collapses cleanly.
+func (l *filerSiblingLister) ListVersions(ctx context.Context, bucket, objectKey string) ([]*filer_pb.Entry, error) {
+	bucketPath := strings.TrimSuffix(l.bucketsPath, "/") + "/" + bucket
+	dir := bucketPath + "/" + objectKey + s3_constants.VersionsFolder
+	const pageSize uint32 = 1024
+	startFrom := ""
+	var versions []*filer_pb.Entry
+	for {
+		var pageCount uint32
+		var lastName string
+		err := filer_pb.SeaweedList(ctx, l.client, dir, "", func(e *filer_pb.Entry, _ bool) error {
+			pageCount++
+			if e == nil {
+				return nil
+			}
+			lastName = e.Name
+			if e.Attributes == nil || e.IsDirectory {
+				return nil
+			}
+			if id, ok := e.Extended[s3_constants.ExtVersionIdKey]; !ok || len(id) == 0 {
+				return nil
+			}
+			versions = append(versions, e)
+			return nil
+		}, startFrom, false, pageSize)
+		if err != nil {
+			if errors.Is(err, filer_pb.ErrNotFound) {
+				return nil, nil
+			}
+			return nil, err
+		}
+		if pageCount < pageSize {
+			return versions, nil
+		}
+		startFrom = lastName
+	}
+}
+
+// LookupNullVersion returns the bare-key entry that represents the
+// null version of objectKey, if any. Both regular files and explicit
+// S3 directory-key markers (an empty directory entry with Mime set)
+// qualify. explicit reports whether the entry's Extended map carries
+// ExtVersionIdKey="null" — the marker the suspended-versioning write
+// path applies. NotFound returns (nil, false, nil).
+func (l *filerSiblingLister) LookupNullVersion(ctx context.Context, bucket, objectKey string) (*filer_pb.Entry, bool, error) {
+	bucketPath := strings.TrimSuffix(l.bucketsPath, "/") + "/" + bucket
+	parent, name := util.NewFullPath(bucketPath, objectKey).DirAndName()
+	resp, err := filer_pb.LookupEntry(ctx, l.client, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: parent,
+		Name:      name,
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return nil, false, nil
+		}
+		return nil, false, err
+	}
+	if resp == nil || resp.Entry == nil {
+		return nil, false, nil
+	}
+	e := resp.Entry
+	if e.IsDirectory && !e.IsDirectoryKeyObject() {
+		return nil, false, nil
+	}
+	explicit := false
+	if id, ok := e.Extended[s3_constants.ExtVersionIdKey]; ok && string(id) == "null" {
+		explicit = true
+	}
+	return e, explicit, nil
+}
+
+// LookupVersion fetches <bucketsPath>/<bucket>/<objectKey>.versions/v_<id>
+// for the pointer-transition router branch. NotFound returns (nil, nil)
+// — the displaced version may have been hard-deleted between the
+// pointer update and the lookup.
+func (l *filerSiblingLister) LookupVersion(ctx context.Context, bucket, objectKey, versionID string) (*filer_pb.Entry, error) {
+	if versionID == "" {
+		return nil, nil
+	}
+	bucketPath := strings.TrimSuffix(l.bucketsPath, "/") + "/" + bucket
+	dir := bucketPath + "/" + objectKey + s3_constants.VersionsFolder
+	resp, err := filer_pb.LookupEntry(ctx, l.client, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: dir,
+		Name:      "v_" + versionID,
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	if resp == nil {
+		return nil, nil
+	}
+	return resp.Entry, nil
+}

weed/s3api/s3lifecycle/reader/reader.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -13,14 +14,36 @@ import (
 )
 
 // Event is one in-shard meta-log event delivered to the router.
+//
+// BootstrapVersion is set only by the bucket bootstrapper when it
+// expands a .versions/ directory; the meta-log path leaves it nil.
+// Carries pre-computed sibling state so the router can fire
+// NoncurrentDays / NewerNoncurrent without listing again.
 type Event struct {
-	TsNs      int64
-	Bucket    string
-	Key       string
-	ShardID   int
-	OldEntry  *filer_pb.Entry
-	NewEntry  *filer_pb.Entry
-	NewParent string
+	TsNs             int64
+	Bucket           string
+	Key              string
+	ShardID          int
+	OldEntry         *filer_pb.Entry
+	NewEntry         *filer_pb.Entry
+	NewParent        string
+	BootstrapVersion *BootstrapVersion
+}
+
+// BootstrapVersion is the per-version state computed once per
+// .versions/<key>/ directory at bootstrap time. Key fields shape
+// EvaluateAction: IsLatest gates current vs. noncurrent rules,
+// NoncurrentIndex gates NewerNoncurrentVersions retention,
+// SuccessorModTime sets the noncurrent clock (when this version was
+// replaced).
+type BootstrapVersion struct {
+	LogicalKey       string
+	VersionID        string
+	IsLatest         bool
+	IsDeleteMarker   bool
+	NumVersions      int
+	NoncurrentIndex  int // 0 = newest noncurrent
+	SuccessorModTime time.Time
 }
 
 // IsDelete reports whether this event removes an entry.

weed/s3api/s3lifecycle/router/router.go

@@ -1,9 +1,13 @@
 package router
 
 import (
+	"context"
+	"sort"
+	"strconv"
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
@@ -11,6 +15,34 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
 )
 
+// SiblingLister inspects the surviving versions of a versioned key.
+// nil receiver or an error means "unknown" — callers suppress. Four
+// queries: Survivors paginates the .versions/ container plus the bare
+// null version (used by sole-survivor and bootstrap); LookupVersion
+// fetches a single version file by id (used by pointer-transition
+// routing to read the displaced version's identity and mtime);
+// ListVersions paginates every version file in the .versions/
+// container (used to compute NoncurrentIndex when a NewerNoncurrent
+// rule is active); LookupNullVersion returns the bare-key entry that
+// represents the null version (used by pointer-transition routing
+// when oldID is empty and to include the null in expansion ranks).
+type SiblingLister interface {
+	Survivors(ctx context.Context, bucket, objectKey string) (Survivors, error)
+	LookupVersion(ctx context.Context, bucket, objectKey, versionID string) (*filer_pb.Entry, error)
+	ListVersions(ctx context.Context, bucket, objectKey string) ([]*filer_pb.Entry, error)
+	LookupNullVersion(ctx context.Context, bucket, objectKey string) (entry *filer_pb.Entry, explicit bool, err error)
+}
+
+// Survivors describes the state under .versions/<key>/ plus the bare
+// null-version that exists when versioning was turned on after the
+// object was first PUT (s3api_object_versioning.go treats <bucket>/<key>
+// as a regular file, the null version, in that case).
+type Survivors struct {
+	Count          int             // entries under .versions/<key>/, capped at 2
+	LoneEntry      *filer_pb.Entry // populated when Count == 1
+	HasNullVersion bool            // bare <bucket>/<key> exists as a regular file
+}
+
 // Match is one (event, action) pair where EvaluateAction fired. The
 // dispatcher runs `LifecycleDelete` at DueTime; identity-CAS in the RPC
 // guards against drift between schedule time and dispatch time.
@@ -42,26 +74,52 @@ type EntryIdentity struct {
 	ExtendedHash []byte
 }
 
-// Route returns the matches that fire for ev against snap. Only EVENT_DRIVEN
-// actions on this bucket are considered; actions in SCAN_AT_DATE or DISABLED
-// modes are out-of-band of the event stream. Inactive actions
-// (BootstrapComplete=false) are also skipped.
-func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match {
+// Route returns the matches that fire for ev against snap. Only active
+// event-driven actions are considered; SCAN_AT_DATE and DISABLED bypass
+// this path.
+func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now time.Time, lister SiblingLister) []Match {
 	if snap == nil || ev == nil {
 		return nil
 	}
-	// Hard deletes carry no schedule-relevant state: an Expiration would
-	// hit NOOP_RESOLVED at dispatch time anyway, ExpiredObjectDeleteMarker
-	// only fires on the latest-version delete-marker which is a Create
-	// from the server's perspective. Skip rather than burn a schedule slot.
-	if ev.NewEntry == nil {
-		return nil
-	}
 	keys := snap.BucketActionKeys(ev.Bucket)
 	if len(keys) == 0 {
 		return nil
 	}
-	info := buildObjectInfo(ev, snap.BucketVersioned(ev.Bucket))
+	// Bootstrap-expanded version event: sibling state is pre-computed,
+	// info.Key is the LOGICAL key so rule prefixes match. Skip the
+	// meta-log path's version-folder skip.
+	if ev.BootstrapVersion != nil {
+		return routeBootstrapVersion(snap, ev, keys)
+	}
+
+	versioned := snap.BucketVersioned(ev.Bucket)
+
+	// .versions/ directory metadata update: when ExtLatestVersionIdKey
+	// changes, the OLD pointer value names a version that's now
+	// noncurrent. Route NoncurrentDays / NewerNoncurrent for it without
+	// waiting for the next bootstrap.
+	if versioned && ev.NewEntry != nil && ev.OldEntry != nil && ev.NewEntry.IsDirectory && isVersionsContainerKey(ev.Key) {
+		if !hasActiveEventDrivenAction(snap, keys, s3lifecycle.ActionKindNoncurrentDays) &&
+			!hasActiveEventDrivenAction(snap, keys, s3lifecycle.ActionKindNewerNoncurrent) {
+			return nil
+		}
+		return routePointerTransition(ctx, snap, ev, keys, lister)
+	}
+
+	// EXP_DM can fire on two version-folder events: the marker create
+	// (sole survivor immediately) and a noncurrent hard-delete that
+	// leaves only the marker behind. Both reach routeSoleSurvivorMarker.
+	if versioned && isVersionFolderPath(ev.Key) {
+		if !hasActiveEventDrivenAction(snap, keys, s3lifecycle.ActionKindExpiredDeleteMarker) {
+			return nil
+		}
+		return routeSoleSurvivorMarker(ctx, snap, ev, keys, lister)
+	}
+
+	if ev.NewEntry == nil {
+		return nil
+	}
+	info := buildObjectInfo(ev, versioned)
 	if info == nil {
 		return nil
 	}
@@ -112,42 +170,540 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match {
 	return matches
 }
 
-// buildObjectInfo derives an ObjectInfo from a meta-log event and the
-// bucket's versioning state. Returns nil when the event has no usable
-// shape (missing attributes, hard delete already handled upstream).
+// routePointerTransition handles a .versions/ container update where
+// ExtLatestVersionIdKey changed: the OLD pointer value names a version
+// that just became noncurrent. Two lookup shapes:
 //
-// On a versioned bucket the storage layout (<key>.versions/v_<id>) is
-// shared between the current latest and the noncurrent versions; the
-// latest pointer lives in the .versions/ directory's Extended map and
-// is updated separately. Without that pointer-transition signal here,
-// the router conservatively classifies every event as if it were the
-// current version (IsLatest=true) so it never deletes the live
-// latest; bootstrap walking + the server-side dispatch guard handle
-// noncurrent retention. NumVersions=0 keeps ExpiredObjectDeleteMarker
-// (which requires sole-survivor) suppressed.
+//   - Pure NoncurrentVersionExpirationDays without NewerNoncurrentVersions:
+//     a single LookupVersion of oldID is enough — the displaced version
+//     is the only one that newly entered eligibility for this rule.
 //
-// MPU init directories at .uploads/<upload_id> populate IsMPUInit and
-// use the destination object key from the entry's Extended map for
-// filter matching, so a rule with Filter.Prefix=foo/ matches an MPU
-// uploading to foo/bar.txt.
+//   - Any active NewerNoncurrentVersions rule: a pointer flip shifts
+//     every prior noncurrent's rank by one, so the version that *just
+//     crossed* the keep-count threshold needs evaluation too. List the
+//     full .versions/ container, rank newest-first, and route every
+//     eligible noncurrent. Identity-CAS handles dedup with earlier
+//     schedules.
+//
+// Without this branch the worker has to wait for the next bootstrap to
+// schedule retention on a freshly-noncurrent version.
+func routePointerTransition(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, lister SiblingLister) []Match {
+	if lister == nil {
+		return nil
+	}
+	logical := strings.TrimSuffix(ev.Key, s3_constants.VersionsFolder)
+	if logical == "" {
+		return nil
+	}
+	oldID := string(ev.OldEntry.Extended[s3_constants.ExtLatestVersionIdKey])
+	newID := string(ev.NewEntry.Extended[s3_constants.ExtLatestVersionIdKey])
+	if oldID == newID {
+		// Same id means the update didn't transition the pointer.
+		return nil
+	}
+	// oldID == "" doesn't mean "nothing displaced": a bare null may
+	// have been the implicit/explicit latest before the pointer
+	// flipped to a real id.
+	// newID == "" means a suspended-versioning write cleared the
+	// pointer and made the bare null current. The cached
+	// ExtLatestVersionMtimeKey may still hold the prior latest's
+	// mtime (stale), so we must NOT use successorModTimeFromContainer
+	// in that case — derive the successor clock from the null entry's
+	// mtime instead. latestIDForExpand carries the same substitution
+	// so the expansion path's latestPos lookup matches the null sibling.
+	var successor time.Time
+	latestIDForExpand := newID
+	if newID == "" {
+		nullEntry, _, err := lister.LookupNullVersion(ctx, ev.Bucket, logical)
+		if err != nil {
+			glog.V(2).Infof("lifecycle router: lookup null %s/%s: %v", ev.Bucket, logical, err)
+			return nil
+		}
+		if nullEntry == nil || nullEntry.Attributes == nil {
+			return nil
+		}
+		successor = time.Unix(nullEntry.Attributes.Mtime, int64(nullEntry.Attributes.MtimeNs))
+		latestIDForExpand = "null"
+	} else {
+		successor = successorModTimeFromContainer(ev.NewEntry)
+	}
+	if successor.IsZero() {
+		return nil
+	}
+	if needsFullExpansion(snap, keys) {
+		return routePointerTransitionExpand(ctx, snap, ev, keys, lister, logical, latestIDForExpand, successor)
+	}
+	return routePointerTransitionDisplaced(ctx, snap, ev, keys, lister, logical, oldID, successor)
+}
+
+// needsFullExpansion reports whether any active event-driven rule on
+// this bucket cares about NoncurrentIndex (NewerNoncurrentVersions > 0
+// in either NoncurrentDays or pure-count NewerNoncurrent).
+func needsFullExpansion(snap *engine.Snapshot, keys []s3lifecycle.ActionKey) bool {
+	for _, k := range keys {
+		if k.ActionKind != s3lifecycle.ActionKindNoncurrentDays && k.ActionKind != s3lifecycle.ActionKindNewerNoncurrent {
+			continue
+		}
+		a := snap.Action(k)
+		if a == nil || !a.IsActive() || a.Mode != engine.ModeEventDriven {
+			continue
+		}
+		if a.Rule != nil && a.Rule.NewerNoncurrentVersions > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+// successorModTimeFromContainer reads the cached latest-version mtime
+// from the .versions/ container's Extended map.
+// updateLatestVersionInDirectory writes it via setCachedListMetadata
+// alongside ExtLatestVersionIdKey, but the directory's own
+// Attributes.Mtime is preserved across pointer updates — using it
+// directly would let a stale dir mtime trigger expiration immediately.
+// Returns zero time if the cached mtime is missing or unparseable; the
+// caller suppresses in that case.
+func successorModTimeFromContainer(entry *filer_pb.Entry) time.Time {
+	raw, ok := entry.Extended[s3_constants.ExtLatestVersionMtimeKey]
+	if !ok || len(raw) == 0 {
+		return time.Time{}
+	}
+	secs, err := strconv.ParseInt(string(raw), 10, 64)
+	if err != nil || secs <= 0 {
+		return time.Time{}
+	}
+	return time.Unix(secs, 0)
+}
+
+// routePointerTransitionDisplaced is the single-lookup path: only the
+// displaced version's noncurrent eligibility could have changed, so
+// fetching just its file is enough. oldID == "" routes the bare null
+// version instead — it was the implicit latest before the pointer
+// flipped to a real id.
+func routePointerTransitionDisplaced(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, lister SiblingLister, logical, oldID string, successor time.Time) []Match {
+	var displaced *filer_pb.Entry
+	displacedID := oldID
+	if oldID == "" {
+		nullEntry, _, err := lister.LookupNullVersion(ctx, ev.Bucket, logical)
+		if err != nil {
+			glog.V(2).Infof("lifecycle router: lookup null version %s/%s: %v", ev.Bucket, logical, err)
+			return nil
+		}
+		if nullEntry == nil {
+			return nil
+		}
+		displaced = nullEntry
+		displacedID = "null"
+	} else {
+		entry, err := lister.LookupVersion(ctx, ev.Bucket, logical, oldID)
+		if err != nil {
+			glog.V(2).Infof("lifecycle router: lookup displaced version %s/%s/%s: %v", ev.Bucket, logical, oldID, err)
+			return nil
+		}
+		displaced = entry
+	}
+	if displaced == nil || displaced.Attributes == nil {
+		return nil
+	}
+	idx := 0
+	info := &s3lifecycle.ObjectInfo{
+		Key:              logical,
+		ModTime:          time.Unix(displaced.Attributes.Mtime, int64(displaced.Attributes.MtimeNs)),
+		Size:             int64(displaced.Attributes.FileSize),
+		IsLatest:         false,
+		IsDeleteMarker:   string(displaced.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
+		NoncurrentIndex:  &idx,
+		SuccessorModTime: successor,
+	}
+	if tags := extractTags(displaced.Extended); len(tags) > 0 {
+		info.Tags = tags
+	}
+	return emitNoncurrentMatches(snap, ev, keys, info, displaced, displacedID, successor)
+}
+
+// routePointerTransitionExpand routes only the versions that newly
+// became eligible by the pointer flip:
+//
+//   - rank 0: the displaced version (newly noncurrent), needed for the
+//     pure-NoncurrentDays clock,
+//   - rank == rule.NewerNoncurrentVersions for each active rule that
+//     gates on count: the version at exactly that rank just crossed
+//     from kept to expired.
+//
+// Emitting every eligible noncurrent on every PUT would push
+// O(versions) heap entries per flip — Schedule.Add doesn't dedup, so
+// identity-CAS at dispatch only stops the wasted RPC, not the heap
+// growth. Bootstrap still owns full backfill.
+func routePointerTransitionExpand(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, lister SiblingLister, logical, newID string, successor time.Time) []Match {
+	rawVersions, err := lister.ListVersions(ctx, ev.Bucket, logical)
+	if err != nil {
+		glog.V(2).Infof("lifecycle router: list versions %s/%s: %v", ev.Bucket, logical, err)
+		return nil
+	}
+	// Include the bare null version in the sibling set: count-based
+	// ranks are wrong if a pre-versioning or suspended-null entry
+	// exists outside .versions/. ID is "null" for sort + emit.
+	nullEntry, _, nullErr := lister.LookupNullVersion(ctx, ev.Bucket, logical)
+	if nullErr != nil {
+		glog.V(2).Infof("lifecycle router: lookup null %s/%s: %v", ev.Bucket, logical, nullErr)
+		return nil
+	}
+	type sibling struct {
+		entry *filer_pb.Entry
+		id    string
+	}
+	siblings := make([]sibling, 0, len(rawVersions)+1)
+	for _, v := range rawVersions {
+		if v == nil || v.Attributes == nil {
+			continue
+		}
+		id := string(v.Extended[s3_constants.ExtVersionIdKey])
+		if id == "" {
+			continue
+		}
+		siblings = append(siblings, sibling{entry: v, id: id})
+	}
+	if nullEntry != nil && nullEntry.Attributes != nil {
+		siblings = append(siblings, sibling{entry: nullEntry, id: "null"})
+	}
+	if len(siblings) == 0 {
+		return nil
+	}
+	sort.SliceStable(siblings, func(i, j int) bool {
+		mi := siblings[i].entry.Attributes.Mtime*int64(1e9) + int64(siblings[i].entry.Attributes.MtimeNs)
+		mj := siblings[j].entry.Attributes.Mtime*int64(1e9) + int64(siblings[j].entry.Attributes.MtimeNs)
+		if mi != mj {
+			return mi > mj
+		}
+		return s3lifecycle.CompareVersionIds(siblings[i].id, siblings[j].id) < 0
+	})
+	// Resolve latestPos by finding newID. Default to -1 so a missing
+	// newID (race with the listing, or torn write) suppresses the
+	// expansion: we'd otherwise call the actual newest sibling "latest"
+	// against the pointer's intent and misrank every noncurrent.
+	// Bootstrap repairs state on the next walk.
+	latestPos := -1
+	if newID != "" {
+		for i, s := range siblings {
+			if s.id == newID {
+				latestPos = i
+				break
+			}
+		}
+	}
+	if latestPos < 0 {
+		glog.V(2).Infof("lifecycle router: pointer transition %s/%s: new id %s not found in listing", ev.Bucket, logical, newID)
+		return nil
+	}
+	noncurrentCount := len(siblings) - 1
+
+	// Collect the target noncurrent ranks: 0 (the freshly displaced)
+	// plus N for each active count-gated rule.
+	rankSet := map[int]struct{}{0: {}}
+	for _, k := range keys {
+		if k.ActionKind != s3lifecycle.ActionKindNoncurrentDays && k.ActionKind != s3lifecycle.ActionKindNewerNoncurrent {
+			continue
+		}
+		a := snap.Action(k)
+		if a == nil || !a.IsActive() || a.Mode != engine.ModeEventDriven {
+			continue
+		}
+		if a.Rule != nil && a.Rule.NewerNoncurrentVersions > 0 {
+			rankSet[a.Rule.NewerNoncurrentVersions] = struct{}{}
+		}
+	}
+	ranks := make([]int, 0, len(rankSet))
+	for r := range rankSet {
+		ranks = append(ranks, r)
+	}
+	sort.Ints(ranks)
+
+	var matches []Match
+	for _, rank := range ranks {
+		if rank >= noncurrentCount {
+			continue
+		}
+		// Convert noncurrent rank to position in the sorted slice,
+		// skipping the latest's slot.
+		i := rank
+		if rank >= latestPos {
+			i = rank + 1
+		}
+		s := siblings[i]
+		// Successor mtime: the entry directly newer than this one in
+		// the sorted list. When the next-newer slot is the latest,
+		// use the cached successor (the new latest's mtime); otherwise
+		// the immediate predecessor's mtime.
+		var thisSuccessor time.Time
+		if i > 0 && i-1 != latestPos {
+			thisSuccessor = time.Unix(siblings[i-1].entry.Attributes.Mtime, int64(siblings[i-1].entry.Attributes.MtimeNs))
+		} else {
+			thisSuccessor = successor
+		}
+		idx := rank
+		info := &s3lifecycle.ObjectInfo{
+			Key:              logical,
+			ModTime:          time.Unix(s.entry.Attributes.Mtime, int64(s.entry.Attributes.MtimeNs)),
+			Size:             int64(s.entry.Attributes.FileSize),
+			IsLatest:         false,
+			IsDeleteMarker:   string(s.entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
+			NoncurrentIndex:  &idx,
+			SuccessorModTime: thisSuccessor,
+			NumVersions:      len(siblings),
+		}
+		if tags := extractTags(s.entry.Extended); len(tags) > 0 {
+			info.Tags = tags
+		}
+		matches = append(matches, emitNoncurrentMatches(snap, ev, keys, info, s.entry, s.id, thisSuccessor)...)
+	}
+	return matches
+}
+
+// emitNoncurrentMatches walks NoncurrentDays / NewerNoncurrent action
+// keys and emits Matches for each one that fires. Shared between the
+// single-lookup and full-expansion paths.
+func emitNoncurrentMatches(snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, info *s3lifecycle.ObjectInfo, entry *filer_pb.Entry, versionID string, successor time.Time) []Match {
+	eventTime := time.Unix(0, ev.TsNs)
+	identity := buildIdentityFromEntry(entry)
+	var matches []Match
+	for _, key := range keys {
+		if key.ActionKind != s3lifecycle.ActionKindNoncurrentDays && key.ActionKind != s3lifecycle.ActionKindNewerNoncurrent {
+			continue
+		}
+		action := snap.Action(key)
+		if action == nil || !action.IsActive() || action.Mode != engine.ModeEventDriven {
+			continue
+		}
+		clock := successor
+		if clock.IsZero() {
+			clock = info.ModTime
+		}
+		dueTime := clock.Add(action.Delay)
+		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
+		if res.Action == s3lifecycle.ActionNone {
+			continue
+		}
+		matches = append(matches, Match{
+			Key:       key,
+			Action:    action,
+			Result:    res,
+			EventTs:   eventTime,
+			DueTime:   dueTime,
+			Bucket:    ev.Bucket,
+			ObjectKey: info.Key,
+			VersionID: versionID,
+			Identity:  identity,
+		})
+	}
+	return matches
+}
+
+// routeSoleSurvivorMarker emits an EXP_DM Match against the LOGICAL key
+// (so the dispatcher can call deleteSpecificObjectVersion) with the
+// marker's version_id. Handles two events: a marker create (the new
+// entry IS the marker) and a noncurrent hard-delete that leaves the
+// marker behind (the listing's lone entry IS the marker). The server
+// re-checks before deleting.
+func routeSoleSurvivorMarker(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, lister SiblingLister) []Match {
+	if lister == nil {
+		return nil
+	}
+	// Skip the listing RPC for events that can't possibly produce a
+	// sole-survivor marker: a regular non-marker version create/update
+	// always lands at Count >= 2.
+	if ev.NewEntry != nil && !isDeleteMarkerEntry(ev.NewEntry) {
+		return nil
+	}
+	logicalKey, ok := logicalKeyFromVersionPath(ev.Key)
+	if !ok {
+		return nil
+	}
+	s, err := lister.Survivors(ctx, ev.Bucket, logicalKey)
+	if err != nil {
+		glog.V(2).Infof("lifecycle router: survivors %s/%s: %v", ev.Bucket, logicalKey, err)
+		return nil
+	}
+	// Pre-versioning bare-key objects (the "null" version) live outside
+	// .versions/. Treating count==1 as sole-survivor while a null
+	// version exists would let lifecycle delete the marker and re-expose
+	// the old object.
+	if s.Count != 1 || s.HasNullVersion || s.LoneEntry == nil {
+		return nil
+	}
+	if !isDeleteMarkerEntry(s.LoneEntry) {
+		return nil
+	}
+	versionID := string(s.LoneEntry.Extended[s3_constants.ExtVersionIdKey])
+	if versionID == "" {
+		// Empty version_id would BLOCK at dispatch and freeze the cursor.
+		return nil
+	}
+	entry := s.LoneEntry
+	info := &s3lifecycle.ObjectInfo{
+		Key:            logicalKey,
+		ModTime:        time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)),
+		Size:           int64(entry.Attributes.FileSize),
+		IsLatest:       true,
+		IsDeleteMarker: true,
+		NumVersions:    1,
+	}
+	if tags := extractTags(entry.Extended); len(tags) > 0 {
+		info.Tags = tags
+	}
+	eventTime := time.Unix(0, ev.TsNs)
+	identity := buildIdentityFromEntry(entry)
+	var matches []Match
+	for _, key := range keys {
+		if key.ActionKind != s3lifecycle.ActionKindExpiredDeleteMarker {
+			continue
+		}
+		action := snap.Action(key)
+		if action == nil || !action.IsActive() || action.Mode != engine.ModeEventDriven {
+			continue
+		}
+		dueTime := info.ModTime.Add(action.Delay)
+		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
+		if res.Action == s3lifecycle.ActionNone {
+			continue
+		}
+		matches = append(matches, Match{
+			Key:       key,
+			Action:    action,
+			Result:    res,
+			EventTs:   eventTime,
+			DueTime:   dueTime,
+			Bucket:    ev.Bucket,
+			ObjectKey: logicalKey,
+			VersionID: versionID,
+			Identity:  identity,
+		})
+	}
+	return matches
+}
+
+// routeBootstrapVersion handles a synthesized event from BucketBootstrapper.
+// The bootstrap walker has already listed .versions/<key>/, sorted siblings
+// newest-first, and stamped each one's IsLatest / NoncurrentIndex /
+// SuccessorModTime. The router only needs to assemble ObjectInfo and run
+// the match loop with the standard kind gates. ev.NewEntry is the version
+// file itself; ev.Key is the version-folder path; the LOGICAL key from
+// BootstrapVersion drives prefix matching and the dispatcher.
+func routeBootstrapVersion(snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey) []Match {
+	bv := ev.BootstrapVersion
+	entry := ev.NewEntry
+	if entry == nil || entry.Attributes == nil || bv.LogicalKey == "" {
+		return nil
+	}
+	idx := bv.NoncurrentIndex
+	info := &s3lifecycle.ObjectInfo{
+		Key:              bv.LogicalKey,
+		ModTime:          time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)),
+		Size:             int64(entry.Attributes.FileSize),
+		IsLatest:         bv.IsLatest,
+		IsDeleteMarker:   bv.IsDeleteMarker,
+		NumVersions:      bv.NumVersions,
+		SuccessorModTime: bv.SuccessorModTime,
+	}
+	if !bv.IsLatest {
+		info.NoncurrentIndex = &idx
+	}
+	if tags := extractTags(entry.Extended); len(tags) > 0 {
+		info.Tags = tags
+	}
+	eventTime := time.Unix(0, ev.TsNs)
+	identity := buildIdentityFromEntry(entry)
+	var matches []Match
+	for _, key := range keys {
+		action := snap.Action(key)
+		if action == nil || !action.IsActive() || action.Mode != engine.ModeEventDriven {
+			continue
+		}
+		// ABORT_MPU never applies to a versioned object.
+		if key.ActionKind == s3lifecycle.ActionKindAbortMPU {
+			continue
+		}
+		// Noncurrent rules clock from when this version was replaced
+		// (SuccessorModTime), not from when it was originally written.
+		// Bootstrap populates SuccessorModTime; fall back to ModTime
+		// for the latest version (no successor exists).
+		clock := info.ModTime
+		if !info.IsLatest && !info.SuccessorModTime.IsZero() {
+			clock = info.SuccessorModTime
+		}
+		dueTime := clock.Add(action.Delay)
+		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
+		if res.Action == s3lifecycle.ActionNone {
+			continue
+		}
+		// Pin the version_id only for kinds the dispatcher needs to
+		// target by version: noncurrent retention and the marker
+		// itself. EXPIRATION_DAYS / EXPIRATION_DATE on the latest
+		// must NOT carry it — between schedule and dispatch a fresh
+		// PUT can land, and identity-CAS against the original
+		// version's bytes would still pass even though the latest has
+		// moved on. Empty VersionID makes the dispatcher fetch the
+		// current latest, where identity-CAS resolves to STALE_IDENTITY
+		// and bootstrap re-schedules with the new latest's identity.
+		var matchVersionID string
+		switch key.ActionKind {
+		case s3lifecycle.ActionKindNoncurrentDays,
+			s3lifecycle.ActionKindNewerNoncurrent,
+			s3lifecycle.ActionKindExpiredDeleteMarker:
+			matchVersionID = bv.VersionID
+		}
+		matches = append(matches, Match{
+			Key:       key,
+			Action:    action,
+			Result:    res,
+			EventTs:   eventTime,
+			DueTime:   dueTime,
+			Bucket:    ev.Bucket,
+			ObjectKey: bv.LogicalKey,
+			VersionID: matchVersionID,
+			Identity:  identity,
+		})
+	}
+	return matches
+}
+
+// logicalKeyFromVersionPath extracts <logical> from <logical>.versions/<file>.
+// Returns false for a bucket-root marker (path = ".versions/<file>"), since
+// AWS S3 has no concept of an object at the bucket key itself.
+func logicalKeyFromVersionPath(versionPath string) (string, bool) {
+	lastSlash := strings.LastIndex(versionPath, "/")
+	if lastSlash <= 0 {
+		return "", false
+	}
+	parent := versionPath[:lastSlash]
+	if !strings.HasSuffix(parent, s3_constants.VersionsFolder) {
+		return "", false
+	}
+	logical := strings.TrimSuffix(parent, s3_constants.VersionsFolder)
+	if logical == "" {
+		return "", false
+	}
+	return logical, true
+}
+
+// buildObjectInfo derives an ObjectInfo from a meta-log event. Returns
+// nil for shapes the router can't classify safely: missing attributes,
+// non-MPU directories, version-folder files (those route through
+// routeSoleSurvivorMarker upstream when EXP_DM applies). On a versioned
+// bucket the latest pointer lives in the .versions/ directory's
+// Extended map; without it we leave NumVersions=0 so the bootstrap walk
+// drives noncurrent retention.
 func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo {
 	entry := ev.NewEntry
 	if entry == nil || entry.Attributes == nil {
 		return nil
 	}
 	if destKey, ok := mpuInitInfo(ev, entry); ok {
-		// MPU intermediate state has no tags, no versions, no delete-marker
-		// semantics. info.Key is the user's destination key so a rule's
-		// Filter.Prefix matches the eventual object; the dispatcher already
-		// carries .uploads/<upload_id> in m.ObjectKey for ABORT_MPU.
 		return &s3lifecycle.ObjectInfo{
 			Key:       destKey,
 			ModTime:   time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)),
 			IsMPUInit: true,
 		}
 	}
-	// Directory entries that aren't MPU inits aren't lifecycle subjects:
-	// the .versions/ folder itself, prefix dirs, etc. — emit nothing.
 	if entry.IsDirectory {
 		return nil
 	}
@@ -159,22 +715,9 @@ func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo {
 		NumVersions: 1,
 	}
 	if versioned {
-		// On a versioned bucket the actual file path doesn't tell us
-		// whether the entry is the current latest or a noncurrent
-		// version — the latest pointer lives in the .versions/
-		// directory's Extended map and isn't part of this event. We
-		// also can't compute NumVersions / NoncurrentIndex here. Skip
-		// any version-folder file event for now; bootstrap walking
-		// drives noncurrent retention and current-version expiration
-		// for versioned buckets until pointer-transition routing
-		// lands. The bare-key path (null-version, pre-versioning
-		// objects) keeps the regular routing.
 		if isVersionFolderPath(ev.Key) {
 			return nil
 		}
-		// NumVersions=0 keeps ExpiredObjectDeleteMarker (sole-survivor
-		// gate) suppressed for the bare-key delete-marker case until
-		// sibling listing lands.
 		info.NumVersions = 0
 	}
 	if tags := extractTags(entry.Extended); len(tags) > 0 {
@@ -186,6 +729,18 @@ func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo {
 	return info
 }
 
+// isVersionsContainerKey reports whether the bucket-relative key IS a
+// .versions/ container itself (e.g. "logs/foo.versions"), as opposed to
+// a file inside one. Used to recognize directory-level events whose
+// Extended map carries the latest pointer for an object.
+func isVersionsContainerKey(key string) bool {
+	if key == s3_constants.VersionsFolder {
+		// Bucket-root .versions: no logical object key.
+		return false
+	}
+	return strings.HasSuffix(key, s3_constants.VersionsFolder)
+}
+
 // isVersionFolderPath reports whether the bucket-relative key sits inside a
 // .versions/ folder — i.e. the path's parent segment ends with the
 // VersionsFolder suffix. Used by the versioned-bucket gate so the router
@@ -226,7 +781,10 @@ func mpuInitInfo(ev *reader.Event, entry *filer_pb.Entry) (destKey string, ok bo
 // buildIdentity captures the entry's schedule-time fingerprint for the CAS
 // witness. Returns nil if the event has no entry to fingerprint (deletes).
 func buildIdentity(ev *reader.Event) *EntryIdentity {
-	entry := ev.NewEntry
+	return buildIdentityFromEntry(ev.NewEntry)
+}
+
+func buildIdentityFromEntry(entry *filer_pb.Entry) *EntryIdentity {
 	if entry == nil {
 		return nil
 	}
@@ -266,10 +824,31 @@ func extractTags(ext map[string][]byte) map[string]string {
 	return out
 }
 
+// hasActiveEventDrivenAction gates I/O (e.g. sibling listing) on whether
+// a match could actually fire. Mirrors the per-key filter in Route so
+// disabled or scan-only actions don't pay the RPC.
+func hasActiveEventDrivenAction(snap *engine.Snapshot, keys []s3lifecycle.ActionKey, kind s3lifecycle.ActionKind) bool {
+	for _, k := range keys {
+		if k.ActionKind != kind {
+			continue
+		}
+		a := snap.Action(k)
+		if a == nil {
+			continue
+		}
+		if a.IsActive() && a.Mode == engine.ModeEventDriven {
+			return true
+		}
+	}
+	return false
+}
+
+// isDeleteMarkerEntry mirrors every read site for ExtDeleteMarkerKey:
+// production writes []byte("true").
 func isDeleteMarkerEntry(entry *filer_pb.Entry) bool {
 	if entry == nil || len(entry.Extended) == 0 {
 		return false
 	}
 	v, ok := entry.Extended[s3_constants.ExtDeleteMarkerKey]
-	return ok && len(v) == 1 && v[0] == 1
+	return ok && string(v) == "true"
 }

weed/s3api/s3lifecycle/scheduler/bootstrap.go

@@ -3,14 +3,18 @@ package scheduler
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 // EventInjector is the bootstrap-side hook into the dispatcher pipeline.
@@ -20,6 +24,46 @@ type EventInjector interface {
 	InjectEvent(ctx context.Context, ev *reader.Event) error
 }
 
+// listPageSize is the page size for paginated directory listings during
+// the bucket walk. The filer caps SeaweedList(..., limit=0) at
+// DirListingLimit (1000 by default) per call, so a single-page list
+// would silently truncate large directories — a correctness bug for
+// noncurrent retention since older versions past the page boundary
+// would never reach the rank/sort math. Atomic so tests can shrink it
+// without racing the async bootstrap goroutines other tests leave
+// behind (KickOffNew dispatches walks via `go b.walkBucket(...)`,
+// and a fresh test's Cleanup might land before those goroutines exit).
+var listPageSize atomic.Uint32
+
+func init() {
+	listPageSize.Store(1024)
+}
+
+// listAll issues paginated SeaweedList calls until the listing is
+// exhausted, invoking fn for every entry. Pagination uses
+// startFrom = lastEntryName (exclusive) to advance.
+func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, fn func(*filer_pb.Entry) error) error {
+	pageSize := listPageSize.Load()
+	startFrom := ""
+	for {
+		var pageCount uint32
+		var lastName string
+		if err := filer_pb.SeaweedList(ctx, client, dir, "", func(e *filer_pb.Entry, _ bool) error {
+			pageCount++
+			if e != nil {
+				lastName = e.Name
+			}
+			return fn(e)
+		}, startFrom, false, pageSize); err != nil {
+			return err
+		}
+		if pageCount < pageSize {
+			return nil
+		}
+		startFrom = lastName
+	}
+}
+
 // BucketBootstrapper backfills already-existing entries when a freshly-PUT
 // rule's bucket appears in the engine. The reader-driven path only sees
 // meta-log events created after the rule lands; without this walk,
@@ -34,33 +78,70 @@ type EventInjector interface {
 // Synthesized events carry TsNs=0 so dispatcher.advance is a no-op for
 // them — the reader still resumes from its persisted cursor on restart.
 //
-// Bucket completion is in-memory per process; a fresh worker run walks
-// each bucket once on first refresh.
+// Walk state is tracked in-memory per process via two maps:
+//   - lastCompleted: time of the last successful walk, drives the
+//     BootstrapInterval cadence; zero interval means "walk once per
+//     process".
+//   - inFlight: buckets whose walk goroutines are still running.
+//     Skipped regardless of cadence so a walk that takes longer than
+//     BootstrapInterval can't trigger a duplicate goroutine on the
+//     next refresh.
 type BucketBootstrapper struct {
 	FilerClient filer_pb.SeaweedFilerClient
 	BucketsPath string
 	Injector    EventInjector
 
-	mu    sync.Mutex
-	known map[string]bool
+	// BootstrapInterval gates re-walks. Zero means "walk once per
+	// process". Non-zero means "walk again once it's been at least
+	// this long since the last completed walk" — the cadence scan_only
+	// actions rely on, since they can only fire from bootstrap.
+	BootstrapInterval time.Duration
+
+	// Now overrides time.Now for tests.
+	Now func() time.Time
+
+	mu            sync.Mutex
+	lastCompleted map[string]time.Time
+	inFlight      map[string]bool
+}
+
+func (b *BucketBootstrapper) now() time.Time {
+	if b.Now != nil {
+		return b.Now()
+	}
+	return time.Now()
 }
 
 // KickOffNew launches a one-shot walker goroutine for every bucket
-// in `buckets` that hasn't been seen before.
+// that's not currently in flight and either has never completed a walk
+// or whose last successful walk finished more than BootstrapInterval ago.
 func (b *BucketBootstrapper) KickOffNew(ctx context.Context, buckets []string) {
 	if b.Injector == nil {
 		return
 	}
+	now := b.now()
 	b.mu.Lock()
-	if b.known == nil {
-		b.known = map[string]bool{}
+	if b.lastCompleted == nil {
+		b.lastCompleted = map[string]time.Time{}
+	}
+	if b.inFlight == nil {
+		b.inFlight = map[string]bool{}
 	}
 	fresh := make([]string, 0, len(buckets))
 	for _, bucket := range buckets {
-		if b.known[bucket] {
+		if b.inFlight[bucket] {
+			// Walk still running from an earlier KickOffNew — never
+			// double up regardless of cadence. A large bucket that
+			// takes longer than BootstrapInterval would otherwise
+			// have a fresh goroutine fire on every refresh tick.
 			continue
 		}
-		b.known[bucket] = true
+		if last, ok := b.lastCompleted[bucket]; ok {
+			if b.BootstrapInterval <= 0 || now.Sub(last) < b.BootstrapInterval {
+				continue
+			}
+		}
+		b.inFlight[bucket] = true
 		fresh = append(fresh, bucket)
 	}
 	b.mu.Unlock()
@@ -75,7 +156,26 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 	root := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket
 	glog.V(0).Infof("lifecycle bootstrap: starting walk for bucket %s (root=%s)", bucket, root)
 	count := 0
-	if err := walkBucketDir(ctx, b.FilerClient, root, root, func(entry *filer_pb.Entry, key string) error {
+	// skipBare records bucket-relative bare-key paths that
+	// expandVersionsDir already routed as the null version. Without it
+	// the walker's regular emission would also fire for the bare entry
+	// — in a versioned bucket buildObjectInfo classifies it as
+	// IsLatest=true, NumVersions=0, and ExpirationDays would create a
+	// stray delete marker that hides the real latest.
+	skipBare := map[string]bool{}
+	var cb func(entry *filer_pb.Entry, key string) error
+	cb = func(entry *filer_pb.Entry, key string) error {
+		if isVersionsDir(entry) {
+			n, err := b.expandVersionsDir(ctx, bucket, root, key, entry, cb, skipBare)
+			count += n
+			return err
+		}
+		if !entry.IsDirectory && skipBare[key] {
+			return nil
+		}
+		if entry.IsDirectoryKeyObject() && skipBare[key] {
+			return nil
+		}
 		ev := &reader.Event{
 			// TsNs=0 sentinel: dispatcher.advance treats <=0 as no-op,
 			// so the reader's persisted cursor isn't ratcheted forward
@@ -88,51 +188,267 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 		}
 		count++
 		return b.Injector.InjectEvent(ctx, ev)
-	}); err != nil {
+	}
+	walkErr := walkBucketDir(ctx, b.FilerClient, root, root, cb)
+	b.mu.Lock()
+	delete(b.inFlight, bucket)
+	if walkErr == nil {
+		// Stamp completion so BootstrapInterval cadence measures from
+		// end-of-walk. Failures leave lastCompleted alone, so the next
+		// KickOffNew sees no record and walks the bucket again.
+		if b.lastCompleted == nil {
+			b.lastCompleted = map[string]time.Time{}
+		}
+		b.lastCompleted[bucket] = b.now()
+	}
+	b.mu.Unlock()
+	if walkErr != nil {
 		if ctx.Err() == nil {
-			glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, err)
+			glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, walkErr)
 		}
 		return
 	}
 	glog.V(0).Infof("lifecycle bootstrap: bucket %s injected %d entries", bucket, count)
 }
 
-// walkBucketDir lists every file under dir recursively and invokes cb
-// with the filer entry plus its bucket-relative key. MPU init dirs at
-// .uploads/<id> are emitted as a single (directory-shaped) entry so the
-// router's MPU detection fires; deeper directories recurse.
-func walkBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot string, cb func(entry *filer_pb.Entry, key string) error) error {
+// versionItem is the per-sibling state expandVersionsDir builds: the
+// filer entry plus its version_id (or "null" for the bare null version
+// living outside .versions/). isExplicitNull marks bare entries the
+// suspended-versioning write path tagged with ExtVersionIdKey="null"
+// (s3api_object_handlers_put.go); we trust those as latest when the
+// .versions/ pointer is missing. A pre-versioning bare object has no
+// such marker, so a missing pointer there is a race window with a new
+// version write and we keep the newest-sibling fallback.
+type versionItem struct {
+	entry          *filer_pb.Entry
+	versionID      string
+	bareKey        string // bucket-relative path; non-empty only for the null version
+	isExplicitNull bool
+}
+
+// expandVersionsDir lists <root>/<key>/ and, when the children look like
+// SeaweedFS version files, injects one reader.Event per version with
+// BootstrapVersion populated. The bare logical key (the "null" version,
+// living outside .versions/) is included as a sibling so:
+//   - pre-versioning objects with a newer .versions/ history fire
+//     NoncurrentDays as id="null"
+//   - suspended-bucket writes (which clear the .versions/ latest pointer)
+//     correctly classify null as the current version while every
+//     .versions/ child becomes noncurrent
+//
+// versionsKey is the bucket-relative path of the .versions/ directory
+// (e.g. "logs/foo.versions"). When the bare null version is included,
+// its bucket-relative path is added to skipBare so the walker's regular
+// emission for the same entry is suppressed.
+//
+// When no child has ExtVersionIdKey the directory is a coincidentally-
+// named user folder; recurse via fallback (the bucket walk's own cb).
+func (b *BucketBootstrapper) expandVersionsDir(ctx context.Context, bucket, root, versionsKey string, versionsEntry *filer_pb.Entry, fallback func(*filer_pb.Entry, string) error, skipBare map[string]bool) (int, error) {
+	logical := strings.TrimSuffix(versionsKey, s3_constants.VersionsFolder)
+	if logical == "" {
+		return 0, nil
+	}
+	versionsDir := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket + "/" + versionsKey
+	// Collect file children only. Subdirectories under .versions/ would
+	// corrupt sort/rank math; the disambiguation pass below also wants
+	// to see only file-shaped children. Paginate so a hot key with
+	// thousands of versions doesn't truncate at DirListingLimit.
 	var children []*filer_pb.Entry
-	if err := filer_pb.SeaweedList(ctx, client, dir, "", func(e *filer_pb.Entry, _ bool) error {
-		children = append(children, e)
+	if err := listAll(ctx, b.FilerClient, versionsDir, func(e *filer_pb.Entry) error {
+		if e != nil && e.Attributes != nil && !e.IsDirectory {
+			children = append(children, e)
+		}
 		return nil
-	}, "", false, 0); err != nil {
+	}); err != nil {
+		return 0, fmt.Errorf("list %s: %w", versionsDir, err)
+	}
+	items := make([]versionItem, 0, len(children)+1)
+	for _, e := range children {
+		if id, ok := e.Extended[s3_constants.ExtVersionIdKey]; ok && len(id) > 0 {
+			items = append(items, versionItem{entry: e, versionID: string(id)})
+		}
+	}
+	if len(items) == 0 {
+		// Coincidentally-named user folder (or an empty .versions
+		// container). fallback is the bucket walk's own cb so nested
+		// .versions/ entries inside still expand.
+		if fallback == nil {
+			return 0, nil
+		}
+		if err := walkBucketDir(ctx, b.FilerClient, versionsDir, root, fallback); err != nil {
+			return 0, err
+		}
+		return 0, nil
+	}
+	// Look up the bare null version. SeaweedFS keeps it at the logical
+	// path for pre-versioning objects and for suspended-bucket writes.
+	// Both shapes count: regular file (PUT'd object) and explicit S3
+	// directory-key marker (object name ends in /).
+	if nullEntry, nullKey, explicit, ok := b.lookupNullVersion(ctx, bucket, logical); ok {
+		items = append(items, versionItem{
+			entry:          nullEntry,
+			versionID:      "null",
+			bareKey:        nullKey,
+			isExplicitNull: explicit,
+		})
+	}
+	// Sort newest-first: primary by mtime ns, fallback by version_id
+	// (CompareVersionIds returns <0 when first arg is newer). PUTs only
+	// set second-level Mtime, so collisions in the same second are
+	// resolved by the canonical version-id ordering used elsewhere.
+	sort.SliceStable(items, func(i, j int) bool {
+		mi := items[i].entry.Attributes.Mtime*int64(1e9) + int64(items[i].entry.Attributes.MtimeNs)
+		mj := items[j].entry.Attributes.Mtime*int64(1e9) + int64(items[j].entry.Attributes.MtimeNs)
+		if mi != mj {
+			return mi > mj
+		}
+		return s3lifecycle.CompareVersionIds(items[i].versionID, items[j].versionID) < 0
+	})
+	// Resolve latest position.
+	//   1. Pointer names a real id -> that wins (in-order or backdated).
+	//   2. Pointer absent + items[0] is an EXPLICIT null (suspended write
+	//      cleared the pointer and tagged the bare object as null, AND
+	//      the bare object is newest by mtime) -> null is latest.
+	//   3. Pointer absent in any other shape: fall back to newest
+	//      sibling. Catches the post-suspended re-enable race window —
+	//      a fresh .versions/<v1> write whose pointer update hasn't
+	//      landed yet outranks the older suspended-null bare object.
+	latestID := string(versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey])
+	latestPos := 0
+	if latestID != "" {
+		for i, it := range items {
+			if it.versionID == latestID {
+				latestPos = i
+				break
+			}
+		}
+	} else if len(items) > 0 && items[0].versionID == "null" && items[0].isExplicitNull {
+		latestPos = 0
+	}
+	count := 0
+	for i, it := range items {
+		var successor time.Time
+		if i > 0 {
+			prev := items[i-1].entry.Attributes
+			successor = time.Unix(prev.Mtime, int64(prev.MtimeNs))
+		}
+		bv := &reader.BootstrapVersion{
+			LogicalKey:       logical,
+			VersionID:        it.versionID,
+			IsLatest:         i == latestPos,
+			IsDeleteMarker:   string(it.entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
+			NumVersions:      len(items),
+			SuccessorModTime: successor,
+		}
+		if !bv.IsLatest {
+			rank := i
+			if i > latestPos {
+				rank = i - 1
+			}
+			bv.NoncurrentIndex = rank
+		}
+		// Event Key for bookkeeping: real version files keep the
+		// .versions/<file> path; the null version uses its bare path
+		// so the dispatcher's identity check resolves to the same
+		// entry the walker would have emitted.
+		evKey := versionsKey + "/" + it.entry.Name
+		if it.versionID == "null" {
+			evKey = it.bareKey
+		}
+		ev := &reader.Event{
+			TsNs:             0,
+			Bucket:           bucket,
+			Key:              evKey,
+			ShardID:          s3lifecycle.ShardID(bucket, logical),
+			NewEntry:         it.entry,
+			BootstrapVersion: bv,
+		}
+		if err := b.Injector.InjectEvent(ctx, ev); err != nil {
+			return count, err
+		}
+		if it.versionID == "null" && skipBare != nil {
+			skipBare[it.bareKey] = true
+		}
+		count++
+	}
+	return count, nil
+}
+
+// lookupNullVersion returns the bare-key entry that represents the null
+// version of logical, if any. Both regular files and S3 directory-key
+// markers (an empty directory entry with Mime set) qualify. The
+// explicit return reports whether the entry's Extended map carries
+// ExtVersionIdKey == "null" — the marker the suspended-versioning
+// write path applies (s3api_object_handlers_put.go). bucketRelKey is
+// the bucket-relative path the walker would otherwise emit, so the
+// caller can suppress the duplicate.
+func (b *BucketBootstrapper) lookupNullVersion(ctx context.Context, bucket, logical string) (entry *filer_pb.Entry, bucketRelKey string, explicit bool, ok bool) {
+	bucketPath := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket
+	parent, name := util.NewFullPath(bucketPath, logical).DirAndName()
+	resp, err := filer_pb.LookupEntry(ctx, b.FilerClient, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: parent,
+		Name:      name,
+	})
+	if err != nil || resp == nil || resp.Entry == nil {
+		return nil, "", false, false
+	}
+	e := resp.Entry
+	if e.IsDirectory && !e.IsDirectoryKeyObject() {
+		return nil, "", false, false
+	}
+	if id, hasID := e.Extended[s3_constants.ExtVersionIdKey]; hasID && string(id) == "null" {
+		explicit = true
+	}
+	return e, strings.TrimPrefix(parent+"/"+name, bucketPath+"/"), explicit, true
+}
+
+// walkBucketDir streams entries under dir and invokes cb. Two kinds of
+// directories are emitted whole rather than recursed into:
+//   - .uploads/<id> MPU init dirs (router fires ABORT_MPU off the dir entry)
+//   - <key>.versions/ directories (caller expands them into per-version
+//     events; recursing here would emit individual version files without
+//     the sibling state needed for NoncurrentDays / NewerNoncurrent)
+//
+// .versions/ dirs are processed before everything else at each level so
+// the cb's expandVersionsDir call can record the bare null-version key
+// in the walk-shared skip set before the same level emits the bare entry.
+// Two streaming passes (rather than buffering the whole directory) trade
+// a second listing for bounded memory on flat buckets with millions of
+// entries.
+func walkBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot string, cb func(entry *filer_pb.Entry, key string) error) error {
+	// Pass 1: .versions/ dirs only.
+	if err := listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
+		if e == nil || e.Attributes == nil {
+			return nil
+		}
+		if !e.IsDirectory || !isVersionsDir(e) {
+			return nil
+		}
+		full := dir + "/" + e.Name
+		key := strings.TrimPrefix(full, bucketRoot+"/")
+		return cb(e, key)
+	}); err != nil {
 		return fmt.Errorf("list %s: %w", dir, err)
 	}
-	for _, entry := range children {
-		if entry == nil || entry.Attributes == nil {
-			continue
+	// Pass 2: everything else. Bare entries whose name was claimed by
+	// a sibling .versions/ expansion are dropped by the cb's skip-set.
+	return listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
+		if e == nil || e.Attributes == nil {
+			return nil
 		}
-		full := dir + "/" + entry.Name
+		if e.IsDirectory && isVersionsDir(e) {
+			return nil
+		}
+		full := dir + "/" + e.Name
 		key := strings.TrimPrefix(full, bucketRoot+"/")
-
-		if entry.IsDirectory {
-			if isMPUInitDir(key, entry) {
-				if err := cb(entry, key); err != nil {
-					return err
-				}
-				continue
+		if e.IsDirectory {
+			if isMPUInitDir(key, e) {
+				return cb(e, key)
 			}
-			if err := walkBucketDir(ctx, client, full, bucketRoot, cb); err != nil {
-				return err
-			}
-			continue
+			return walkBucketDir(ctx, client, full, bucketRoot, cb)
 		}
-		if err := cb(entry, key); err != nil {
-			return err
-		}
-	}
-	return nil
+		return cb(e, key)
+	})
 }
 
 // isMPUInitDir mirrors router.mpuInitInfo: a directory at .uploads/<id>
@@ -152,3 +468,16 @@ func isMPUInitDir(key string, entry *filer_pb.Entry) bool {
 	return ok && len(v) > 0
 }
 
+// isVersionsDir matches `<x>.versions/` by name suffix. We can't gate on
+// ExtLatestVersionIdKey here: createDeleteMarker writes the version file
+// before updating the parent's Extended pointer, so a walk that races
+// with that update would see the directory without the pointer and
+// recurse into raw version files, losing the sibling state needed for
+// noncurrent rules. expandVersionsDir handles disambiguation by
+// inspecting children for ExtVersionIdKey; coincidentally-named
+// directories that aren't real .versions storage fall through to a
+// regular recursion.
+func isVersionsDir(entry *filer_pb.Entry) bool {
+	return entry.IsDirectory && strings.HasSuffix(entry.Name, s3_constants.VersionsFolder)
+}
+

weed/s3api/s3lifecycle/scheduler/scheduler.go

@@ -33,11 +33,12 @@ type Scheduler struct {
 	ClientID    int32
 	ClientName  string
 
-	Workers         int
-	DispatchTick    time.Duration
-	CheckpointTick  time.Duration
-	RefreshInterval time.Duration
-	RetryBackoff    time.Duration
+	Workers           int
+	DispatchTick      time.Duration
+	CheckpointTick    time.Duration
+	RefreshInterval   time.Duration
+	BootstrapInterval time.Duration
+	RetryBackoff      time.Duration
 }
 
 // Run blocks until ctx is canceled. Spawns Workers + 1 goroutines: one
@@ -100,9 +101,10 @@ func (s *Scheduler) Run(ctx context.Context) error {
 	}
 
 	bs := &BucketBootstrapper{
-		FilerClient: s.FilerClient,
-		BucketsPath: s.BucketsPath,
-		Injector:    pipelineFanout(pipelinesByShard),
+		FilerClient:       s.FilerClient,
+		BucketsPath:       s.BucketsPath,
+		Injector:          pipelineFanout(pipelinesByShard),
+		BootstrapInterval: s.BootstrapInterval,
 	}
 
 	s.refreshEngine(ctx, bs)

weed/s3api/s3lifecycle/version_time.go

@@ -1,6 +1,76 @@
 package s3lifecycle
 
+import (
+	"strconv"
+)
+
 // 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
+
+// CompareVersionIds returns negative if a is newer than b, positive if b
+// is newer, 0 if equal. Mirrors compareVersionIds in s3api_version_id.go
+// (kept duplicated to avoid the s3api -> s3lifecycle import cycle). Used
+// as a tiebreak when version mtimes collide at second resolution.
+func CompareVersionIds(a, b string) int {
+	if a == b {
+		return 0
+	}
+	if a == "null" {
+		return 1 // null sorts last
+	}
+	if b == "null" {
+		return -1
+	}
+	aIsNew := isNewFormatVersionId(a)
+	bIsNew := isNewFormatVersionId(b)
+	if aIsNew == bIsNew {
+		// Same format. New format (inverted timestamps) sorts smaller=newer
+		// lexicographically; old format sorts smaller=older.
+		if aIsNew {
+			if a < b {
+				return -1
+			}
+			return 1
+		}
+		if a < b {
+			return 1
+		}
+		return -1
+	}
+	at := getVersionTimestamp(a)
+	bt := getVersionTimestamp(b)
+	if at > bt {
+		return -1
+	}
+	if at < bt {
+		return 1
+	}
+	return 0
+}
+
+func isNewFormatVersionId(versionId string) bool {
+	if len(versionId) < 16 || versionId == "null" {
+		return false
+	}
+	t, err := strconv.ParseUint(versionId[:16], 16, 64)
+	if err != nil {
+		return false
+	}
+	return t > versionIdFormatThreshold
+}
+
+func getVersionTimestamp(versionId string) int64 {
+	if len(versionId) < 16 || versionId == "null" {
+		return 0
+	}
+	t, err := strconv.ParseUint(versionId[:16], 16, 64)
+	if err != nil {
+		return 0
+	}
+	if t > versionIdFormatThreshold {
+		return int64(^uint64(0)>>1) - int64(t)
+	}
+	return int64(t)
+}

weed/shell/command_s3_lifecycle_run_shard.go

@@ -68,6 +68,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i
 	dispatchTick := fs.Duration("dispatch", 5*time.Second, "dispatcher tick cadence")
 	checkpointTick := fs.Duration("checkpoint", 30*time.Second, "cursor checkpoint cadence")
 	refreshInterval := fs.Duration("refresh", 5*time.Minute, "interval for rebuilding the engine snapshot from filer-backed bucket configs; 0 = compile once at startup")
+	bootstrapInterval := fs.Duration("bootstrap-interval", 0, "cadence for revisiting each bucket's bootstrap walk; 0 = walk once per process. scan_only actions only fire from bootstrap, so a long-running worker needs a non-zero value to handle their retention horizon")
 	runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout. -events alone can hang on quiet shards")
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -119,9 +120,10 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i
 		}
 
 		bsr := &scheduler.BucketBootstrapper{
-			FilerClient: filerClient,
-			BucketsPath: bucketsPath,
-			Injector:    pipeline,
+			FilerClient:       filerClient,
+			BucketsPath:       bucketsPath,
+			Injector:          pipeline,
+			BootstrapInterval: *bootstrapInterval,
 		}
 		bootstrapCtx, bootstrapCancel := context.WithCancel(context.Background())
 		defer bootstrapCancel()

weed/storage/store.go

@@ -394,12 +394,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	collectionVolumeDeletedBytes := make(map[string]int64)
 	collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
 	for _, location := range s.Locations {
-		// keepRemoteData is parallel to deleteVids: true entries preserve the
-		// cloud-tier object on Volume.Destroy. IO-error deletions on a
-		// remote-tiered volume must not nuke the remote object — the error
-		// is local/transient and the cloud copy is the source of truth.
 		var deleteVids []needle.VolumeId
-		var keepRemoteData []bool
 		effectiveMaxCount := location.MaxVolumeCount
 		if location.isDiskSpaceLow {
 			usedSlots := int32(location.LocalVolumesLen())
@@ -420,26 +415,42 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 			if maxFileKey < curMaxFileKey {
 				maxFileKey = curMaxFileKey
 			}
-			shouldDeleteVolume := false
 
-			if v.lastIoError != nil {
-				deleteVids = append(deleteVids, v.Id)
-				keepRemoteData = append(keepRemoteData, v.HasRemoteFile())
-				shouldDeleteVolume = true
-				glog.Warningf("volume %d has IO error: %v", v.Id, v.lastIoError)
+			ioErr, ioCount, quarantined := v.getIoErrorState()
+			if quarantined || (ioErr != nil && ioCount >= IoErrorTolerance) {
+				// Sustained EIO: stop announcing this replica so the master
+				// re-replicates from healthy peers, and mark it read-only
+				// so further writes fail fast instead of producing more
+				// EIOs. Never physically delete the data — the disk may
+				// be transiently bad and this could be the last good
+				// copy. The quarantine is sticky: a stray successful read
+				// clears the streak counter but must not silently put a
+				// known-bad replica back into rotation; recovery is via
+				// MarkVolumeWritable.
+				if !quarantined {
+					glog.Warningf("volume %d quarantined after %d consecutive IO errors: %v",
+						v.Id, ioCount, ioErr)
+					v.markIoQuarantined()
+				}
+				v.noWriteLock.Lock()
+				v.noWriteOrDelete = true
+				v.noWriteLock.Unlock()
+				// Skip per-volume size and read-only bookkeeping: a
+				// quarantined replica should not be summed into the
+				// collection's reported total nor counted in the
+				// read-only stats.
+				continue
+			}
+
+			shouldDeleteVolume := false
+			if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
+				volumeMessages = append(volumeMessages, volumeMessage)
 			} else {
-				if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
-					volumeMessages = append(volumeMessages, volumeMessage)
+				if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
+					deleteVids = append(deleteVids, v.Id)
+					shouldDeleteVolume = true
 				} else {
-					if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
-						if !shouldDeleteVolume {
-							deleteVids = append(deleteVids, v.Id)
-							keepRemoteData = append(keepRemoteData, false)
-							shouldDeleteVolume = true
-						}
-					} else {
-						glog.V(0).Infof("volume %d is expired", v.Id)
-					}
+					glog.V(0).Infof("volume %d is expired", v.Id)
 				}
 			}
 
@@ -483,8 +494,8 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		if len(deleteVids) > 0 {
 			// delete expired volumes.
 			location.volumesLock.Lock()
-			for i, vid := range deleteVids {
-				found, err := location.deleteVolumeById(vid, false, keepRemoteData[i])
+			for _, vid := range deleteVids {
+				found, err := location.deleteVolumeById(vid, false, false)
 				if err == nil {
 					if found {
 						glog.V(0).Infof("volume %d is deleted", vid)
@@ -678,6 +689,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 	v.noWriteOrDelete = false
 	v.PersistReadOnly(false)
 	v.noWriteLock.Unlock()
+	// Clear the EIO streak and the sticky quarantine flag so the next
+	// CollectHeartbeat can announce the volume again. If the disk is
+	// still bad, the next failed op will re-arm the streak.
+	v.resetIoErrorState()
 	return nil
 }
 

weed/storage/volume.go

@@ -53,7 +53,77 @@ type Volume struct {
 	location         *DiskLocation
 	diskId           uint32 // ID of this volume's disk in Store.Locations array
 
-	lastIoError error
+	// lastIoError is the most recent EIO from a read/write/delete; cleared
+	// on the next successful or non-EIO op. lastIoErrorCount tracks
+	// consecutive EIOs so CollectHeartbeat can require a sustained failure
+	// before unmounting the replica — protects against a transient
+	// hardware/network blip hitting multiple replicas at once and
+	// stranding the only good copy.
+	//
+	// ioErrorQuarantined is sticky: once CollectHeartbeat sees the streak
+	// cross IoErrorTolerance it sets this and never clears it on its own.
+	// A subsequent successful read clears the streak counter but must NOT
+	// un-quarantine the volume — only MarkVolumeWritable does that, after
+	// an operator has decided the disk is healthy. Without the sticky
+	// bit, one good read between heartbeats would silently put a known-
+	// bad replica back into rotation.
+	//
+	// All four fields are guarded together so the heartbeat reader sees
+	// a consistent snapshot.
+	lastIoError        error
+	lastIoErrorCount   int32
+	ioErrorQuarantined bool
+	lastIoErrorLock    sync.RWMutex
+}
+
+// noteIoError records an EIO and increments the consecutive-error
+// counter. Caller has already verified errors.Is(err, syscall.EIO).
+func (v *Volume) noteIoError(err error) {
+	v.lastIoErrorLock.Lock()
+	defer v.lastIoErrorLock.Unlock()
+	v.lastIoError = err
+	v.lastIoErrorCount++
+}
+
+// clearIoError resets the EIO streak counter only. The sticky quarantine
+// bit set by CollectHeartbeat is intentionally left alone — recovery is
+// an operator decision via MarkVolumeWritable. Called on any successful
+// op or on a non-EIO error (which still breaks the EIO streak; only
+// sustained EIOs are diagnostic of a failing volume).
+func (v *Volume) clearIoError() {
+	v.lastIoErrorLock.Lock()
+	defer v.lastIoErrorLock.Unlock()
+	v.lastIoError = nil
+	v.lastIoErrorCount = 0
+}
+
+// resetIoErrorState clears both the EIO streak and the sticky quarantine
+// flag. Used by MarkVolumeWritable to rejoin a previously-quarantined
+// replica; if the disk is still bad, the next failed op re-arms the
+// streak.
+func (v *Volume) resetIoErrorState() {
+	v.lastIoErrorLock.Lock()
+	defer v.lastIoErrorLock.Unlock()
+	v.lastIoError = nil
+	v.lastIoErrorCount = 0
+	v.ioErrorQuarantined = false
+}
+
+// markIoQuarantined sets the sticky quarantine flag. Idempotent; safe
+// to call from CollectHeartbeat each pass while the volume remains
+// quarantined.
+func (v *Volume) markIoQuarantined() {
+	v.lastIoErrorLock.Lock()
+	defer v.lastIoErrorLock.Unlock()
+	v.ioErrorQuarantined = true
+}
+
+// getIoErrorState returns the latest EIO, the consecutive-EIO count,
+// and the sticky quarantine flag as one consistent snapshot.
+func (v *Volume) getIoErrorState() (error, int32, bool) {
+	v.lastIoErrorLock.RLock()
+	defer v.lastIoErrorLock.RUnlock()
+	return v.lastIoError, v.lastIoErrorCount, v.ioErrorQuarantined
 }
 
 func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) {

weed/storage/volume_io_error_test.go

@@ -0,0 +1,210 @@
+package storage
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+// eioBackend is a backend.BackendStorageFile that returns EIO from
+// every ReadAt — used to verify the streaming read path threads
+// errors through checkReadWriteError.
+type eioBackend struct{}
+
+func (eioBackend) ReadAt(p []byte, off int64) (int, error) { return 0, syscall.EIO }
+func (eioBackend) WriteAt(p []byte, off int64) (int, error) {
+	return len(p), nil
+}
+func (eioBackend) Truncate(int64) error                          { return nil }
+func (eioBackend) Close() error                                  { return nil }
+func (eioBackend) GetStat() (int64, time.Time, error)            { return 0, time.Time{}, nil }
+func (eioBackend) Name() string                                  { return "eio" }
+func (eioBackend) Sync() error                                   { return nil }
+
+func TestCheckReadWriteErrorTracksConsecutiveEIO(t *testing.T) {
+	v := &Volume{}
+
+	// each EIO bumps the counter.
+	for i := int32(1); i <= 5; i++ {
+		v.checkReadWriteError(fmt.Errorf("disk failed: %w", syscall.EIO))
+		_, count, _ := v.getIoErrorState()
+		if count != i {
+			t.Fatalf("after %d EIO(s): counter = %d, want %d", i, count, i)
+		}
+	}
+
+	// a single success resets the streak (but not the quarantine flag,
+	// which is not set here).
+	v.checkReadWriteError(nil)
+	if err, count, q := v.getIoErrorState(); err != nil || count != 0 || q {
+		t.Fatalf("success did not reset state: err=%v count=%d quarantined=%v", err, count, q)
+	}
+}
+
+func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) {
+	v := &Volume{}
+
+	v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	if _, count, _ := v.getIoErrorState(); count != 2 {
+		t.Fatalf("expected count=2 after two EIOs, got %d", count)
+	}
+
+	// a non-EIO error breaks the streak — only sustained EIOs are
+	// diagnostic of a failing disk.
+	v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC))
+	if err, count, _ := v.getIoErrorState(); err != nil || count != 0 {
+		t.Fatalf("non-EIO did not reset streak: err=%v count=%d", err, count)
+	}
+
+	// a fresh EIO starts the streak from 1, not 3.
+	v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	if _, count, _ := v.getIoErrorState(); count != 1 {
+		t.Fatalf("EIO after non-EIO did not restart streak: count=%d, want 1", count)
+	}
+}
+
+func TestCheckReadWriteErrorIgnoresPlainError(t *testing.T) {
+	v := &Volume{}
+
+	v.checkReadWriteError(errors.New("some other error"))
+	if err, count, _ := v.getIoErrorState(); err != nil || count != 0 {
+		t.Fatalf("non-EIO with no prior streak set state: err=%v count=%d", err, count)
+	}
+}
+
+func TestIoErrorToleranceGate(t *testing.T) {
+	v := &Volume{}
+
+	for i := 0; i < IoErrorTolerance-1; i++ {
+		v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	}
+	if _, count, _ := v.getIoErrorState(); count >= IoErrorTolerance {
+		t.Fatalf("counter %d already crossed tolerance %d after %d errors",
+			count, IoErrorTolerance, IoErrorTolerance-1)
+	}
+
+	v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	if _, count, _ := v.getIoErrorState(); count < IoErrorTolerance {
+		t.Fatalf("counter %d below tolerance %d after %d errors",
+			count, IoErrorTolerance, IoErrorTolerance)
+	}
+}
+
+// Once CollectHeartbeat marks a replica quarantined, a stray successful
+// read must NOT silently put a known-bad disk back into rotation. Only
+// MarkVolumeWritable (resetIoErrorState) clears the sticky bit.
+func TestQuarantineIsSticky(t *testing.T) {
+	v := &Volume{}
+
+	for i := 0; i < IoErrorTolerance; i++ {
+		v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	}
+	v.markIoQuarantined()
+
+	if _, _, q := v.getIoErrorState(); !q {
+		t.Fatalf("markIoQuarantined did not set the flag")
+	}
+
+	// A successful read clears the streak counter…
+	v.checkReadWriteError(nil)
+	if _, count, q := v.getIoErrorState(); count != 0 {
+		t.Fatalf("success did not clear streak: count=%d", count)
+	} else if !q {
+		t.Fatalf("success cleared the sticky quarantine flag — operator-only recovery violated")
+	}
+
+	// …and a non-EIO error also clears the streak but not the flag.
+	v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC))
+	if _, _, q := v.getIoErrorState(); !q {
+		t.Fatalf("non-EIO cleared the sticky quarantine flag")
+	}
+
+	// Only resetIoErrorState (used by MarkVolumeWritable) un-quarantines.
+	v.resetIoErrorState()
+	if err, count, q := v.getIoErrorState(); err != nil || count != 0 || q {
+		t.Fatalf("resetIoErrorState left state: err=%v count=%d quarantined=%v", err, count, q)
+	}
+}
+
+// Streaming/range reads (ReadNeedleBlob) used to bypass the EIO
+// counter, so a failing disk taking range GETs all day would never
+// trip IoErrorTolerance. ReadNeedleBlob now threads the backend error
+// through checkReadWriteError; one EIO must bump the streak by one.
+func TestReadNeedleBlobTracksEIO(t *testing.T) {
+	v := &Volume{
+		DataBackend: eioBackend{},
+		SuperBlock:  super_block.SuperBlock{Version: needle.GetCurrentVersion()},
+		volumeInfo:  &volume_server_pb.VolumeInfo{Version: uint32(needle.GetCurrentVersion())},
+	}
+
+	if _, err := v.ReadNeedleBlob(0, 1); !errors.Is(err, syscall.EIO) {
+		t.Fatalf("expected EIO from fake backend, got %v", err)
+	}
+	if _, count, _ := v.getIoErrorState(); count != 1 {
+		t.Fatalf("ReadNeedleBlob did not bump EIO streak: count=%d", count)
+	}
+}
+
+func TestIoErrorStateIsRaceFree(t *testing.T) {
+	// Relies on `go test -race` to detect any unprotected access on
+	// lastIoError / lastIoErrorCount / ioErrorQuarantined.
+	v := &Volume{}
+
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+			}
+		}
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				v.checkReadWriteError(nil)
+			}
+		}
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				v.markIoQuarantined()
+				v.resetIoErrorState()
+			}
+		}
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			v.getIoErrorState()
+		}
+		close(stop)
+	}()
+	wg.Wait()
+}

weed/storage/volume_read.go

@@ -180,6 +180,14 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
 		if readOption.HasSlowRead {
 			v.dataFileAccessLock.RUnlock()
 		}
+		// Thread the underlying read error through the EIO tracker.
+		// Without this, large/range GETs through readNeedleDataInto
+		// would never trip IoErrorTolerance even on a failing disk.
+		// io.EOF is treated as a clean end-of-stream below, not an
+		// error.
+		if err != nil && err != io.EOF {
+			v.checkReadWriteError(err)
+		}
 
 		toWrite := min(count, int(offset+size-x))
 		if toWrite > 0 {
@@ -201,6 +209,11 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
 			break
 		}
 	}
+	// Whole-needle read completed without a backend error — clear any
+	// pending EIO streak. If a non-EIO failure happens later (CRC etc.)
+	// we still return that error to the caller, but the disk itself
+	// produced clean bytes.
+	v.checkReadWriteError(nil)
 	if offset == 0 && size == int64(n.DataSize) && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) {
 		// the crc.Value() function is to be deprecated. this double checking is for backward compatibility
 		// with seaweed version using crc.Value() instead of uint32(crc), which appears in commit 056c480eb
@@ -224,7 +237,9 @@ func (v *Volume) ReadNeedleBlob(offset int64, size Size) ([]byte, error) {
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()
 
-	return needle.ReadNeedleBlob(v.DataBackend, offset, size, v.Version())
+	blob, err := needle.ReadNeedleBlob(v.DataBackend, offset, size, v.Version())
+	v.checkReadWriteError(err)
+	return blob, err
 }
 
 type VolumeFileScanner interface {

weed/storage/volume_write.go

@@ -17,16 +17,25 @@ var ErrorNotFound = errors.New("not found")
 var ErrorDeleted = errors.New("already deleted")
 var ErrorSizeMismatch = errors.New("size mismatch")
 
+// IoErrorTolerance is the number of consecutive EIOs a volume must
+// see before CollectHeartbeat treats the replica as broken. A single
+// transient error is forgiven so a brief NFS / fabric / power blip
+// affecting several replicas at once does not cascade into removal of
+// the last healthy copy.
+const IoErrorTolerance = 3
+
 func (v *Volume) checkReadWriteError(err error) {
 	if err == nil {
-		if v.lastIoError != nil {
-			v.lastIoError = nil
-		}
+		v.clearIoError()
 		return
 	}
 	if errors.Is(err, syscall.EIO) {
-		v.lastIoError = err
+		v.noteIoError(err)
+		return
 	}
+	// non-EIO error breaks the EIO streak — only sustained EIOs should
+	// be treated as a failing volume.
+	v.clearIoError()
 }
 
 // isFileUnchanged checks whether this needle to write is same as last one.

weed/worker/tasks/s3_lifecycle/config.go

@@ -9,31 +9,34 @@ import (
 const (
 	jobType = "s3_lifecycle"
 
-	defaultWorkers                = 1
-	defaultDispatchTickMinutes    = int64(1)
-	defaultCheckpointTickSeconds  = int64(30)
-	defaultRefreshIntervalMinutes = int64(5)
-	defaultMaxRuntimeMinutes      = int64(60)
+	defaultWorkers                  = 1
+	defaultDispatchTickMinutes      = int64(1)
+	defaultCheckpointTickSeconds    = int64(30)
+	defaultRefreshIntervalMinutes   = int64(5)
+	defaultMaxRuntimeMinutes        = int64(60)
+	defaultBootstrapIntervalMinutes = int64(0) // 0 = walk once per process
 )
 
 // Config is the parsed AdminConfigForm + WorkerConfigForm view.
 type Config struct {
-	Workers         int
-	DispatchTick    time.Duration
-	CheckpointTick  time.Duration
-	RefreshInterval time.Duration
-	MaxRuntime      time.Duration
+	Workers           int
+	DispatchTick      time.Duration
+	CheckpointTick    time.Duration
+	RefreshInterval   time.Duration
+	BootstrapInterval time.Duration
+	MaxRuntime        time.Duration
 }
 
 // ParseConfig pulls the lifecycle Handler config from the merged
 // admin+worker config values. Missing fields fall back to defaults.
 func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Config {
 	cfg := Config{
-		Workers:         int(readInt64(adminValues, "workers", defaultWorkers)),
-		DispatchTick:    time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute,
-		CheckpointTick:  time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second,
-		RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute,
-		MaxRuntime:      time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute,
+		Workers:           int(readInt64(adminValues, "workers", defaultWorkers)),
+		DispatchTick:      time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute,
+		CheckpointTick:    time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second,
+		RefreshInterval:   time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute,
+		BootstrapInterval: time.Duration(readInt64(workerValues, "bootstrap_interval_minutes", defaultBootstrapIntervalMinutes)) * time.Minute,
+		MaxRuntime:        time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute,
 	}
 	if cfg.Workers <= 0 {
 		cfg.Workers = defaultWorkers
@@ -47,6 +50,13 @@ func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Co
 	if cfg.RefreshInterval <= 0 {
 		cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute
 	}
+	// BootstrapInterval is intentionally NOT clamped — zero means
+	// "walk once per process", which is the legacy default for any
+	// deployment that hasn't opted into a cadence yet. Negative values
+	// fall back to zero.
+	if cfg.BootstrapInterval < 0 {
+		cfg.BootstrapInterval = 0
+	}
 	if cfg.MaxRuntime <= 0 {
 		cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute
 	}

weed/worker/tasks/s3_lifecycle/config_test.go

@@ -24,6 +24,9 @@ func TestParseConfigDefaults(t *testing.T) {
 	if cfg.MaxRuntime != 60*time.Minute {
 		t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime)
 	}
+	if cfg.BootstrapInterval != 0 {
+		t.Errorf("BootstrapInterval default=%v, want 0 (walk-once-per-process)", cfg.BootstrapInterval)
+	}
 }
 
 func TestParseConfigOverrides(t *testing.T) {
@@ -31,10 +34,11 @@ func TestParseConfigOverrides(t *testing.T) {
 		"workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}},
 	}
 	worker := map[string]*plugin_pb.ConfigValue{
-		"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}},
-		"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}},
-		"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}},
-		"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
+		"dispatch_tick_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}},
+		"checkpoint_tick_seconds":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}},
+		"refresh_interval_minutes":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}},
+		"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
+		"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
 	}
 	cfg := ParseConfig(admin, worker)
 	if cfg.Workers != 4 {
@@ -49,6 +53,9 @@ func TestParseConfigOverrides(t *testing.T) {
 	if cfg.RefreshInterval != 10*time.Minute {
 		t.Errorf("RefreshInterval=%v, want 10m", cfg.RefreshInterval)
 	}
+	if cfg.BootstrapInterval != 120*time.Minute {
+		t.Errorf("BootstrapInterval=%v, want 120m", cfg.BootstrapInterval)
+	}
 	if cfg.MaxRuntime != 120*time.Minute {
 		t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime)
 	}
@@ -56,10 +63,11 @@ func TestParseConfigOverrides(t *testing.T) {
 
 func TestParseConfigClampsZeroAndNegative(t *testing.T) {
 	worker := map[string]*plugin_pb.ConfigValue{
-		"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
-		"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
-		"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
-		"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
+		"dispatch_tick_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"checkpoint_tick_seconds":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"refresh_interval_minutes":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
+		"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
 	}
 	cfg := ParseConfig(nil, worker)
 	if cfg.DispatchTick != 1*time.Minute {
@@ -71,7 +79,23 @@ func TestParseConfigClampsZeroAndNegative(t *testing.T) {
 	if cfg.RefreshInterval != 5*time.Minute {
 		t.Errorf("zero RefreshInterval should clamp to default, got %v", cfg.RefreshInterval)
 	}
+	if cfg.BootstrapInterval != 0 {
+		t.Errorf("negative BootstrapInterval should clamp to 0, got %v", cfg.BootstrapInterval)
+	}
 	if cfg.MaxRuntime != 60*time.Minute {
 		t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime)
 	}
 }
+
+func TestParseConfigBootstrapIntervalZeroIsLegacy(t *testing.T) {
+	// Explicit zero stays zero (walk-once-per-process). Existing
+	// deployments that don't set bootstrap_interval_minutes get the
+	// legacy behavior unchanged.
+	worker := map[string]*plugin_pb.ConfigValue{
+		"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+	}
+	cfg := ParseConfig(nil, worker)
+	if cfg.BootstrapInterval != 0 {
+		t.Errorf("zero BootstrapInterval must stay zero, got %v", cfg.BootstrapInterval)
+	}
+}

weed/worker/tasks/s3_lifecycle/handler.go

@@ -122,6 +122,14 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
 							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
 						},
+						{
+							Name:        "bootstrap_interval_minutes",
+							Label:       "Bootstrap Re-walk Interval (minutes)",
+							Description: "How often each bucket is re-walked. scan_only rules — those whose retention horizon exceeds meta-log retention — only fire from the bootstrap walk, so a non-zero value is required to enforce them on a long-running worker. 0 keeps the legacy walk-once-per-process behavior.",
+							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64,
+							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
+							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+						},
 						{
 							Name:        "max_runtime_minutes",
 							Label:       "Max Runtime (minutes)",
@@ -134,10 +142,11 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 				},
 			},
 			DefaultValues: map[string]*plugin_pb.ConfigValue{
-				"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}},
-				"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}},
-				"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}},
-				"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}},
+				"dispatch_tick_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}},
+				"checkpoint_tick_seconds":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}},
+				"refresh_interval_minutes":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}},
+				"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultBootstrapIntervalMinutes}},
+				"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}},
 			},
 		},
 		AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
@@ -240,17 +249,18 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ
 	rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn)
 
 	sched := &scheduler.Scheduler{
-		BucketsPath:     bucketsPath,
-		Engine:          engine.New(),
-		Persister:       &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
-		Client:          lifecycleRPCAdapter{c: rpc},
-		FilerClient:     filerClient,
-		ClientID:        util.RandomInt32(),
-		ClientName:      "worker-s3-lifecycle",
-		Workers:         cfg.Workers,
-		DispatchTick:    cfg.DispatchTick,
-		CheckpointTick:  cfg.CheckpointTick,
-		RefreshInterval: cfg.RefreshInterval,
+		BucketsPath:       bucketsPath,
+		Engine:            engine.New(),
+		Persister:         &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
+		Client:            lifecycleRPCAdapter{c: rpc},
+		FilerClient:       filerClient,
+		ClientID:          util.RandomInt32(),
+		ClientName:        "worker-s3-lifecycle",
+		Workers:           cfg.Workers,
+		DispatchTick:      cfg.DispatchTick,
+		CheckpointTick:    cfg.CheckpointTick,
+		RefreshInterval:   cfg.RefreshInterval,
+		BootstrapInterval: cfg.BootstrapInterval,
 	}
 	if err := sched.Run(runCtx); err != nil {
 		glog.Warningf("s3 lifecycle execute: %v", err)