feat(s3/lifecycle): pointer-transition routing for live PUTs (Phase 5b/4) (#9385)

* feat(s3/lifecycle): pointer-transition routing for live PUTs (Phase 5b/4)

Bootstrap covers existing versions, but a live PUT that creates a new
.versions/<v-new> file and updates the parent's ExtLatestVersionIdKey
didn't fire NoncurrentDays / NewerNoncurrent on the displaced prior
version until the next bootstrap. Close that runtime gap.

The meta-log already emits an Update event for the .versions/
directory itself when the latest pointer changes; the router was
dropping it because buildObjectInfo returns nil for directories. New
branch in Route detects that shape (versioned bucket, NewEntry +
OldEntry both directories with the .versions/ suffix, ExtLatestVersionIdKey
changed, ID different from the new ID) and emits a Match against the
LOGICAL key with VersionID=oldID. Match.Identity comes from a single
LookupVersion RPC for the displaced version file; SuccessorModTime is
the directory update's mtime, which is the moment the displaced
version became noncurrent.

SiblingLister grows LookupVersion(bucket, key, versionID) for that
single-RPC fetch. filerSiblingLister implements it; routing path
treats NotFound as "displaced version was hard-deleted in the
meantime, suppress" rather than an error.

The router gates the lookup on at least one active event-driven
NoncurrentDays / NewerNoncurrent rule for the bucket, so most buckets
pay nothing per directory update.

Tests: pointer-flip fires NoncurrentDays with displaced version_id;
unchanged pointer skips; empty old pointer skips (first-PUT scenario);
displaced-version NotFound suppresses; no-rule skips lookup;
NewerNoncurrentVersions retains rank-0; unversioned bucket skips.

* fix(s3/lifecycle): SuccessorModTime cache + NewerNoncurrent expansion

Two correctness gaps in pointer-transition routing.

The .versions/ directory's own Attributes.Mtime is preserved across
pointer updates by updateLatestVersionInDirectory: it's a stale clock
relative to the freshly-written latest version. Using it as the
displaced version's SuccessorModTime made NoncurrentDays compute
due = staleMtime + days, which fires immediately on a fresh PUT
into an old .versions/ container. Read ExtLatestVersionMtimeKey
written by setCachedListMetadata; suppress (return no matches) when
the cache is missing rather than fall back to dir mtime.

Single-oldID lookup is only enough for pure
NoncurrentVersionExpirationDays. Any rule with NewerNoncurrentVersions
> 0 cares about the noncurrent ranks, and a pointer flip shifts every
prior noncurrent's index by one — the version that just crossed the
keep-count threshold needs to be evaluated too. When any matching
rule needs ranks, list the full .versions/ container, sort newest-
first with mtime + version-id tiebreak, and route every noncurrent
with its real index. Identity-CAS dedups against earlier schedules.

SiblingLister grows ListVersions(bucket, key); filerSiblingLister's
implementation paginates the container fully.

Two regression tests: stale dir mtime + correct cached mtime
schedules ~30 days out (not immediate); NewerNoncurrentVersions=2
with 4 versions fires on the rank-2 entry that just crossed the
threshold while rank-0/1 are retained.

* fix(s3/lifecycle): bound pointer-transition expansion to threshold crossings

routePointerTransitionExpand emitted a Match for every eligible
noncurrent on every PUT. Schedule.Add doesn't dedup, so identity-CAS
at dispatch only saved the wasted RPC, not the heap slot. A hot key
with many already-eligible versions and a count rule would push
O(versions) entries per flip, repeatedly, until dispatch caught up.

Bound the emission to versions that newly entered eligibility on
this specific flip: rank 0 (the displaced version, for the
NoncurrentDays clock) plus rank == rule.NewerNoncurrentVersions
for each active count-gated rule (the version that just crossed
from kept to expired). Bootstrap still owns full backfill for
versions that were already over-threshold.

Adds a regression with 6 versions and NewerNoncurrentVersions=2:
asserts only the rank-2 entry that just crossed fires, not the
already-over-threshold rank-3/rank-4 entries.

* fix(s3/lifecycle): suppress pointer-transition expansion when newID missing

routePointerTransitionExpand defaulted latestPos to 0 if newID wasn't
found in the listing. That made the actual newest sibling latest
against the pointer's intent, then misranked every other version. A
race between the pointer write and the version write could land us
there.

Default latestPos to -1, set it only on a real match, and suppress
the expansion when the search misses. Bootstrap repairs state on
the next walk.

The NewerNoncurrentVersions retention test was setting only
lookupEntry, so Route never reached the expansion path it claimed to
exercise. Repoint to listVersions and assert ListVersions was
consulted while LookupVersion was not. Adds a regression covering
the missing-newID suppression directly.

* fix(s3/lifecycle): include bare null version in pointer-transition routing

Bootstrap models the bare-key object as a "null" sibling alongside
.versions/ children, but the live pointer-transition path didn't.
Two cases lost:

1. oldID == "" was treated as "nothing displaced". A pre-versioning
   bare object becomes noncurrent when the first versioned PUT lands
   and the pointer flips to a real id, but live routing skipped it
   and waited for the next bootstrap.

2. The expansion path's ListVersions returned only .versions/
   children. With a bare null in the picture, the noncurrent ranks
   were wrong, so NewerNoncurrentVersions could keep the wrong
   versions and delete the right ones (or vice versa).

SiblingLister grows LookupNullVersion(bucket, key) returning the
bare entry plus an explicit-null flag (matches the bootstrap shape).
filerSiblingLister implements it via util.NewFullPath +
filer_pb.LookupEntry.

routePointerTransitionDisplaced: oldID == "" now consults
LookupNullVersion. When the bare entry exists, route it as
VersionID="null" against the LOGICAL key.

routePointerTransitionExpand: collect .versions/ children plus the
null entry into one sibling slice before sorting and ranking. The
threshold-crossing logic now sees the same N-version set that
bootstrap would compute.

Three new tests: oldID == "" with no null is a no-op (one null
lookup, no version lookup); oldID == "" with a bare null schedules
NoncurrentDays as VersionID="null"; expansion with a bare null
between .versions/ siblings places null at its mtime-correct rank
and only that rank-N entry fires.

* fix(s3/lifecycle): atomic listPageSize so test cleanup doesn't race

KickOffNew dispatches walks via `go b.walkBucket(...)`. A test that
finishes before its goroutines drain leaves them running into the
next test's t.Cleanup, which mutates listPageSize. -race spots the
read/write collision intermittently. Convert listPageSize to
atomic.Uint32; tests use Load/Store. No production semantics change.

* fix(s3/lifecycle): null becomes latest when suspended PUT clears pointer

The router treated newID == "" as if the cached
ExtLatestVersionMtimeKey were still authoritative — but that cache
holds the displaced version's mtime, written by setCachedListMetadata
when the prior version became latest. Using it as SuccessorModTime
made NoncurrentDays=30 immediately fire on a 100-day-old displaced
version even though it became noncurrent today.

When newID == "" the bare null is the new latest. Look it up,
substitute its mtime as the successor clock, and substitute "null"
as the latestPos target for the expansion path's id match. Both
displaced and expand paths now derive the right clock.

updateIsLatestFlagsForSuspendedVersioning was the upstream cause of
the staleness — it cleared ExtLatestVersionIdKey and FileNameKey but
left the cached size/mtime/etag/owner/delete-marker behind. Call
clearCachedVersionMetadata so the .versions/ container is consistent
with "null is latest". The router-side guard is still needed for
older deployments that ran the buggy code, but new writes won't
exercise the workaround.

Two regressions: 100-day-old displaced under NoncurrentDays=30 with
a today-null PUT schedules ~30d out (not immediate); same shape with
NewerNoncurrentVersions=2 ranks the null at latest and only the
rank-2 entry fires.

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/sibling_lister.go

@@ -59,3 +59,104 @@ func (l *filerSiblingLister) Survivors(ctx context.Context, bucket, objectKey st
 	}
 	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/router/router.go

@@ -2,6 +2,8 @@ package router
 
 import (
 	"context"
+	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -14,9 +16,21 @@ import (
 )
 
 // SiblingLister inspects the surviving versions of a versioned key.
-// nil receiver or an error means "unknown" — callers suppress.
+// 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
@@ -80,6 +94,18 @@ func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now tim
 
 	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.
@@ -144,6 +170,334 @@ func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now tim
 	return matches
 }
 
+// routePointerTransition handles a .versions/ container update where
+// ExtLatestVersionIdKey changed: the OLD pointer value names a version
+// that just became noncurrent. Two lookup shapes:
+//
+//   - 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.
+//
+//   - 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
@@ -375,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

weed/s3api/s3lifecycle/router/router_test.go

@@ -3,6 +3,7 @@ package router
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -499,9 +500,34 @@ func markerEvent(bucket, logicalKey, versionID string, mtimeUnix, mtimeNs int64)
 // state to return; calls list is appended on each invocation so tests
 // can assert whether the lister was consulted at all.
 type recordingLister struct {
-	calls    []string
-	survivors Survivors
-	err      error
+	calls           []string
+	survivors       Survivors
+	err             error
+	lookupCalls     []string
+	lookupEntry     *filer_pb.Entry
+	lookupErr       error
+	listCalls       []string
+	listVersions    []*filer_pb.Entry
+	listErr         error
+	nullCalls       []string
+	nullEntry       *filer_pb.Entry
+	nullExplicit    bool
+	nullErr         error
+}
+
+func (r *recordingLister) ListVersions(_ context.Context, bucket, key string) ([]*filer_pb.Entry, error) {
+	r.listCalls = append(r.listCalls, bucket+"/"+key)
+	return r.listVersions, r.listErr
+}
+
+func (r *recordingLister) LookupNullVersion(_ context.Context, bucket, key string) (*filer_pb.Entry, bool, error) {
+	r.nullCalls = append(r.nullCalls, bucket+"/"+key)
+	return r.nullEntry, r.nullExplicit, r.nullErr
+}
+
+func (r *recordingLister) LookupVersion(_ context.Context, bucket, key, versionID string) (*filer_pb.Entry, error) {
+	r.lookupCalls = append(r.lookupCalls, bucket+"/"+key+"@"+versionID)
+	return r.lookupEntry, r.lookupErr
 }
 
 func (r *recordingLister) Survivors(_ context.Context, bucket, key string) (Survivors, error) {
@@ -992,3 +1018,636 @@ func TestRouteBootstrapVersionAbortMPUNeverEmittedForVersion(t *testing.T) {
 		t.Fatalf("bootstrap version event must not produce ABORT_MPU match, got %v", got)
 	}
 }
+
+// versionsContainerEvent builds a .versions/<key>/ directory update.
+// The NEW entry carries the cached latest-version mtime (the value
+// setCachedListMetadata writes alongside the latest pointer). The
+// directory's own Mtime is preserved at containerStaleMtime so the
+// router can't accidentally use it as the successor clock.
+func versionsContainerEvent(bucket, logical, oldID, newID string, latestMtimeUnix int64) *reader.Event {
+	const containerStaleMtime int64 = 1
+	mk := func(id string, includeMtime bool) *filer_pb.Entry {
+		ext := map[string][]byte{}
+		if id != "" {
+			ext[s3_constants.ExtLatestVersionIdKey] = []byte(id)
+		}
+		if includeMtime {
+			ext[s3_constants.ExtLatestVersionMtimeKey] = []byte(fmt.Sprintf("%d", latestMtimeUnix))
+		}
+		return &filer_pb.Entry{
+			Name:        logical + s3_constants.VersionsFolder,
+			IsDirectory: true,
+			Attributes:  &filer_pb.FuseAttributes{Mtime: containerStaleMtime},
+			Extended:    ext,
+		}
+	}
+	return &reader.Event{
+		Bucket:   bucket,
+		Key:      logical + s3_constants.VersionsFolder,
+		OldEntry: mk(oldID, false),
+		NewEntry: mk(newID, true),
+	}
+}
+
+func displacedVersionEntry(versionID string, mtimeUnix int64) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: "v_" + versionID,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:    mtimeUnix,
+			FileSize: 100,
+		},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte(versionID),
+		},
+	}
+}
+
+func TestRoutePointerTransitionFiresNoncurrentDays(t *testing.T) {
+	// .versions/<key>/ directory update flips ExtLatestVersionIdKey from
+	// v-old to v-new. v-old becomes noncurrent immediately. The router
+	// looks up v-old's file (one RPC) and emits NoncurrentDays Match
+	// against the LOGICAL key with VersionID=v-old.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	displaced := displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix())
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+
+	lister := &recordingLister{lookupEntry: displaced}
+	matches := Route(context.Background(), snap, ev, now, lister)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (NoncurrentDays), got %v", matches)
+	}
+	m := matches[0]
+	if m.ObjectKey != "logs/foo" {
+		t.Fatalf("ObjectKey=%q, want logs/foo", m.ObjectKey)
+	}
+	if m.VersionID != "v-old" {
+		t.Fatalf("VersionID=%q, want displaced v-old", m.VersionID)
+	}
+	if len(lister.lookupCalls) != 1 || lister.lookupCalls[0] != "bk/logs/foo@v-old" {
+		t.Fatalf("lookup calls=%v, want [bk/logs/foo@v-old]", lister.lookupCalls)
+	}
+}
+
+func TestRoutePointerTransitionPointerUnchangedSkipped(t *testing.T) {
+	// Directory update with same OLD/NEW pointer (e.g. some other
+	// metadata changed): no transition; nothing to schedule and no
+	// lookup RPC.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-same", "v-same", now.Unix())
+
+	lister := &recordingLister{}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("unchanged pointer must not fire, got %v", got)
+	}
+	if len(lister.lookupCalls) != 0 {
+		t.Fatalf("unchanged pointer must not consult lister: %v", lister.lookupCalls)
+	}
+}
+
+func TestRoutePointerTransitionEmptyOldPointerNoNullSkipped(t *testing.T) {
+	// First PUT on a brand-new versioned object: OLD pointer is empty,
+	// no bare null version exists. Nothing displaced.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "", "v-new", now.Unix())
+
+	lister := &recordingLister{} // nullEntry is nil
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("empty old pointer + no null must not fire, got %v", got)
+	}
+	if len(lister.lookupCalls) != 0 {
+		t.Fatalf("LookupVersion must not be called for empty oldID, got %v", lister.lookupCalls)
+	}
+	if len(lister.nullCalls) != 1 {
+		t.Fatalf("expected exactly one LookupNullVersion call, got %v", lister.nullCalls)
+	}
+}
+
+func TestRoutePointerTransitionEmptyOldPointerWithNullSchedules(t *testing.T) {
+	// First versioned PUT after a pre-versioning bare object exists.
+	// OLD pointer is empty, but the bare null is the displaced version.
+	// NoncurrentDays must schedule it as VersionID="null" so the worker
+	// doesn't have to wait for the next bootstrap.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "", "v-new", now.Unix())
+	nullMt := now.AddDate(0, 0, -10)
+	lister := &recordingLister{nullEntry: &filer_pb.Entry{
+		Name: "foo",
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:    nullMt.Unix(),
+			FileSize: 50,
+		},
+	}}
+	matches := Route(context.Background(), snap, ev, now, lister)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (NoncurrentDays on null), got %v", matches)
+	}
+	if matches[0].VersionID != "null" {
+		t.Fatalf("VersionID=%q, want \"null\"", matches[0].VersionID)
+	}
+	if matches[0].ObjectKey != "logs/foo" {
+		t.Fatalf("ObjectKey=%q, want logs/foo", matches[0].ObjectKey)
+	}
+}
+
+func TestRoutePointerTransitionExpansionIncludesNullVersion(t *testing.T) {
+	// Suspended-bucket history: bare null exists with a recent mtime.
+	// Versioning re-enabled and a new version was just written. With
+	// NewerNoncurrentVersions=2 the rank-2 entry is the threshold-
+	// crosser. Because the null sits between v-mid and v-old by mtime,
+	// it occupies a noncurrent rank slot and shifts what the rank-2
+	// entry actually IS. Without including null, ranks would be wrong.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-cur", "v-new", now.Unix())
+	// Mtimes (newest-first): v-new, v-cur, v-mid, null, v-old, v-old2.
+	// Post-flip ranks: latest=v-new, rank0=v-cur, rank1=v-mid,
+	// rank2=null, rank3=v-old, rank4=v-old2. The rank-2 crossing is
+	// "null" — and without including null in the sibling set, rank2
+	// would have been v-old (different version_id, wrong identity).
+	listVersions := []*filer_pb.Entry{
+		displacedVersionEntry("v-new", now.Unix()),
+		displacedVersionEntry("v-cur", now.AddDate(0, 0, -1).Unix()),
+		displacedVersionEntry("v-mid", now.AddDate(0, 0, -2).Unix()),
+		displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix()),
+		displacedVersionEntry("v-old2", now.AddDate(0, 0, -20).Unix()),
+	}
+	nullEntry := &filer_pb.Entry{
+		Name:       "foo",
+		Attributes: &filer_pb.FuseAttributes{Mtime: now.AddDate(0, 0, -3).Unix()},
+	}
+	lister := &recordingLister{listVersions: listVersions, nullEntry: nullEntry}
+
+	matches := Route(context.Background(), snap, ev, now, lister)
+	versionIDs := []string{}
+	for _, m := range matches {
+		versionIDs = append(versionIDs, m.VersionID)
+	}
+	if !contains(versionIDs, "null") {
+		t.Fatalf("rank-2 should be null after including bare entry, matches=%v", versionIDs)
+	}
+	for _, id := range []string{"v-cur", "v-mid", "v-old", "v-old2", "v-new"} {
+		if contains(versionIDs, id) {
+			t.Fatalf("only the rank-2 (null) entry should fire, got %s in matches=%v", id, versionIDs)
+		}
+	}
+}
+
+func TestRoutePointerTransitionDisplacedVersionMissingSuppressed(t *testing.T) {
+	// Race: by the time the router looks up v-old, it's already been
+	// hard-deleted. LookupVersion returns (nil, nil); no Match.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+
+	lister := &recordingLister{lookupEntry: nil} // not found
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("missing displaced version must suppress, got %v", got)
+	}
+	if len(lister.lookupCalls) != 1 {
+		t.Fatalf("lookup attempted once, got %v", lister.lookupCalls)
+	}
+}
+
+func TestRoutePointerTransitionNoNoncurrentRuleSkipsLookup(t *testing.T) {
+	// Bucket has only ExpirationDays — no NoncurrentDays / NewerNoncurrent.
+	// The router must NOT issue the lookup RPC.
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+
+	lister := &recordingLister{lookupEntry: displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix())}
+	Route(context.Background(), snap, ev, now, lister)
+	if len(lister.lookupCalls) != 0 {
+		t.Fatalf("lister consulted without noncurrent rule: %v", lister.lookupCalls)
+	}
+}
+
+func TestRoutePointerTransitionNewerNoncurrentNewestNoncurrentRetained(t *testing.T) {
+	// NewerNoncurrentVersions=2 routes through the expansion path. The
+	// freshly-noncurrent version is at rank 0 (newest noncurrent) and
+	// the threshold-crossing rank N=2 doesn't exist (only 2 versions
+	// total). No match expected — and ListVersions must be the one
+	// consulted, not LookupVersion.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+	lister := &recordingLister{listVersions: []*filer_pb.Entry{
+		displacedVersionEntry("v-new", now.Unix()),
+		displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix()),
+	}}
+
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("rank-0 noncurrent must be retained under NewerNoncurrentVersions=2, got %v", got)
+	}
+	if len(lister.listCalls) != 1 {
+		t.Fatalf("expansion path must consult ListVersions, calls=%v", lister.listCalls)
+	}
+	if len(lister.lookupCalls) != 0 {
+		t.Fatalf("expansion path must not consult LookupVersion, calls=%v", lister.lookupCalls)
+	}
+}
+
+func TestRoutePointerTransitionExpansionMissingNewIDSuppressed(t *testing.T) {
+	// Race window: by the time ListVersions returns, the new pointer's
+	// version file isn't visible yet. latestPos can't resolve, so the
+	// router suppresses (bootstrap repairs state) instead of treating
+	// the actual newest sibling as latest and misranking every other
+	// version.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+	// Listing returns v-old + v-mid but NOT v-new (the just-named latest).
+	lister := &recordingLister{listVersions: []*filer_pb.Entry{
+		displacedVersionEntry("v-mid", now.AddDate(0, 0, -1).Unix()),
+		displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix()),
+	}}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("missing new id must suppress, got %v", got)
+	}
+}
+
+func TestRoutePointerTransitionUnversionedBucketSkipped(t *testing.T) {
+	// Same event shape on an unversioned bucket: should not even reach
+	// the pointer-transition branch.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWith(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+	lister := &recordingLister{}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("unversioned bucket must not route pointer transitions, got %v", got)
+	}
+}
+
+func TestRoutePointerTransitionMissingCachedMtimeSuppressed(t *testing.T) {
+	// Older builds (or buggy paths) may write the latest pointer
+	// without ExtLatestVersionMtimeKey. Without a reliable successor
+	// clock NoncurrentDays would compute a year-0001 base and fire
+	// immediately. Suppress until the cache lands.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	// Hand-build the event WITHOUT ExtLatestVersionMtimeKey on NewEntry.
+	now := time.Now()
+	ev := &reader.Event{
+		Bucket: "bk",
+		Key:    "logs/foo" + s3_constants.VersionsFolder,
+		OldEntry: &filer_pb.Entry{
+			IsDirectory: true,
+			Attributes:  &filer_pb.FuseAttributes{Mtime: 1},
+			Extended:    map[string][]byte{s3_constants.ExtLatestVersionIdKey: []byte("v-old")},
+		},
+		NewEntry: &filer_pb.Entry{
+			IsDirectory: true,
+			Attributes:  &filer_pb.FuseAttributes{Mtime: 1},
+			Extended:    map[string][]byte{s3_constants.ExtLatestVersionIdKey: []byte("v-new")},
+		},
+	}
+	displaced := displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix())
+	lister := &recordingLister{lookupEntry: displaced}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("missing cached mtime must suppress, got %v", got)
+	}
+	if len(lister.lookupCalls) != 0 {
+		t.Fatalf("must not consult lister without successor mtime, got %v", lister.lookupCalls)
+	}
+}
+
+func TestRoutePointerTransitionUsesCachedMtimeNotStaleDirMtime(t *testing.T) {
+	// Regression: the .versions/ directory's own Attributes.Mtime is
+	// preserved across pointer updates by updateLatestVersionInDirectory.
+	// Using it as SuccessorModTime would let a fresh pointer flip on an
+	// old directory fire NoncurrentDays right away. Use the cached
+	// latest-mtime instead.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 30,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	// The cached latest-mtime IS now (fresh PUT). Container's Attrs.Mtime
+	// is stale (containerStaleMtime=1). With the buggy old code, the
+	// match would fire immediately because successor=year-1970 +
+	// 30 days < now. With the fix, due = cached latest mtime + 30 days,
+	// which is in the future.
+	ev := versionsContainerEvent("bk", "logs/foo", "v-old", "v-new", now.Unix())
+	displaced := displacedVersionEntry("v-old", now.AddDate(0, 0, -10).Unix())
+	lister := &recordingLister{lookupEntry: displaced}
+
+	matches := Route(context.Background(), snap, ev, now, lister)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (scheduled, not fired), got %v", matches)
+	}
+	if !matches[0].DueTime.After(now.Add(29 * 24 * time.Hour)) {
+		t.Fatalf("DueTime=%v, want ~30d from now (cached mtime + 30d)", matches[0].DueTime)
+	}
+}
+
+func TestRoutePointerTransitionNewerNoncurrentExpansionFiresOnCrossingThreshold(t *testing.T) {
+	// NewerNoncurrentVersions=2 keeps the 2 newest noncurrents. Before
+	// the pointer flip there were 3 versions: v-cur (latest), v-mid
+	// (rank 0 noncurrent), v-old (rank 1 noncurrent). After flipping
+	// to v-new the ranks shift to: v-new latest, v-cur rank 0, v-mid
+	// rank 1, v-old rank 2 — v-old just crossed the threshold and
+	// must fire NoncurrentDays this run instead of waiting for the
+	// next bootstrap.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	// Successor mtime (cached on container) is "now" — the new latest.
+	ev := versionsContainerEvent("bk", "logs/foo", "v-cur", "v-new", now.Unix())
+	allVersions := []*filer_pb.Entry{
+		displacedVersionEntry("v-new", now.Unix()),                  // newest
+		displacedVersionEntry("v-cur", now.AddDate(0, 0, -1).Unix()),
+		displacedVersionEntry("v-mid", now.AddDate(0, 0, -10).Unix()),
+		displacedVersionEntry("v-old", now.AddDate(0, 0, -30).Unix()), // oldest
+	}
+	lister := &recordingLister{listVersions: allVersions}
+
+	matches := Route(context.Background(), snap, ev, now, lister)
+	// v-cur (rank 0) and v-mid (rank 1) retained; v-old (rank 2) fires.
+	versionIDs := []string{}
+	for _, m := range matches {
+		versionIDs = append(versionIDs, m.VersionID)
+	}
+	if !contains(versionIDs, "v-old") {
+		t.Fatalf("v-old at rank 2 (>= NewerNoncurrentVersions=2) must fire, matches=%v", versionIDs)
+	}
+	for _, id := range []string{"v-cur", "v-mid"} {
+		if contains(versionIDs, id) {
+			t.Fatalf("rank-%d noncurrent must be retained, matches=%v", indexOf([]string{"v-cur", "v-mid"}, id), versionIDs)
+		}
+	}
+	if len(lister.listCalls) != 1 {
+		t.Fatalf("expansion path must call ListVersions once, got %v", lister.listCalls)
+	}
+	if len(lister.lookupCalls) != 0 {
+		t.Fatalf("expansion path must not call LookupVersion, got %v", lister.lookupCalls)
+	}
+}
+
+func contains(ss []string, s string) bool {
+	for _, x := range ss {
+		if x == s {
+			return true
+		}
+	}
+	return false
+}
+
+func indexOf(ss []string, s string) int {
+	for i, x := range ss {
+		if x == s {
+			return i
+		}
+	}
+	return -1
+}
+
+func TestRoutePointerTransitionExpansionEmitsOnlyThresholdCrossing(t *testing.T) {
+	// Hot key with many already-eligible noncurrents under
+	// NewerNoncurrentVersions=2. A pointer flip must NOT enqueue every
+	// over-threshold version (Schedule.Add doesn't dedup; identity-CAS
+	// only saves the dispatch RPC, not the heap slot). Only the version
+	// that JUST crossed from kept to expired needs to enter the heap;
+	// everything else was already scheduled by an earlier transition or
+	// bootstrap.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	ev := versionsContainerEvent("bk", "logs/foo", "v-cur", "v-new", now.Unix())
+	// 6 versions total: v-new latest after flip, then v-cur (rank 0),
+	// v-mid (rank 1), v-2 (rank 2 — newly crossed), v-3, v-4 (already
+	// past threshold from previous flips). Only rank 2 should enter
+	// the heap on this flip.
+	allVersions := []*filer_pb.Entry{
+		displacedVersionEntry("v-new", now.Unix()),
+		displacedVersionEntry("v-cur", now.AddDate(0, 0, -1).Unix()),
+		displacedVersionEntry("v-mid", now.AddDate(0, 0, -2).Unix()),
+		displacedVersionEntry("v-2", now.AddDate(0, 0, -10).Unix()),
+		displacedVersionEntry("v-3", now.AddDate(0, 0, -20).Unix()),
+		displacedVersionEntry("v-4", now.AddDate(0, 0, -30).Unix()),
+	}
+	lister := &recordingLister{listVersions: allVersions}
+
+	matches := Route(context.Background(), snap, ev, now, lister)
+	versionIDs := []string{}
+	for _, m := range matches {
+		versionIDs = append(versionIDs, m.VersionID)
+	}
+	// Want exactly v-2 (rank 2 — newly crossed). Not v-3 / v-4 (already
+	// over) and not v-cur / v-mid (still kept).
+	if !contains(versionIDs, "v-2") {
+		t.Fatalf("v-2 at the new crossing rank must fire, matches=%v", versionIDs)
+	}
+	for _, id := range []string{"v-3", "v-4"} {
+		if contains(versionIDs, id) {
+			t.Fatalf("over-threshold %s must NOT re-enter the heap on this flip, matches=%v", id, versionIDs)
+		}
+	}
+	for _, id := range []string{"v-cur", "v-mid"} {
+		if contains(versionIDs, id) {
+			t.Fatalf("retained %s must not fire, matches=%v", id, versionIDs)
+		}
+	}
+}
+
+// versionsContainerEventStaleMtime simulates the pointer-cleared
+// suspended-versioning shape: NewEntry has no ExtLatestVersionIdKey
+// but DOES carry a stale ExtLatestVersionMtimeKey from the displaced
+// version (the cached value the buggy server-side code left behind).
+func versionsContainerEventStaleMtime(bucket, logical, oldID string, staleMtimeUnix int64) *reader.Event {
+	const containerStaleMtime int64 = 1
+	return &reader.Event{
+		Bucket: bucket,
+		Key:    logical + s3_constants.VersionsFolder,
+		OldEntry: &filer_pb.Entry{
+			Name:        logical + s3_constants.VersionsFolder,
+			IsDirectory: true,
+			Attributes:  &filer_pb.FuseAttributes{Mtime: containerStaleMtime},
+			Extended: map[string][]byte{
+				s3_constants.ExtLatestVersionIdKey:    []byte(oldID),
+				s3_constants.ExtLatestVersionMtimeKey: []byte(fmt.Sprintf("%d", staleMtimeUnix)),
+			},
+		},
+		NewEntry: &filer_pb.Entry{
+			Name:        logical + s3_constants.VersionsFolder,
+			IsDirectory: true,
+			Attributes:  &filer_pb.FuseAttributes{Mtime: containerStaleMtime},
+			Extended: map[string][]byte{
+				// Pointer cleared. Cached mtime intentionally left stale
+				// (the bug we're guarding against on the router side).
+				s3_constants.ExtLatestVersionMtimeKey: []byte(fmt.Sprintf("%d", staleMtimeUnix)),
+			},
+		},
+	}
+}
+
+func TestRoutePointerTransitionSuspendedClearsPointerUsesNullMtime(t *testing.T) {
+	// Suspended write: NewEntry's ExtLatestVersionIdKey is gone. The
+	// cached ExtLatestVersionMtimeKey may still hold the displaced
+	// version's mtime (server-side bug + defensive router behavior).
+	// Router must use the null entry's mtime as the successor clock,
+	// not the cached stale value — otherwise NoncurrentDays=30 fires
+	// on a 100-day-old displaced version even though it became
+	// noncurrent today.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 30,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	staleMtime := now.AddDate(0, 0, -100).Unix() // displaced version is 100 days old
+	ev := versionsContainerEventStaleMtime("bk", "logs/foo", "v-old", staleMtime)
+	displaced := displacedVersionEntry("v-old", staleMtime)
+	// Null was written today (the suspended PUT itself).
+	nullEntry := &filer_pb.Entry{
+		Name:       "foo",
+		Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix()},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte("null")},
+	}
+	lister := &recordingLister{lookupEntry: displaced, nullEntry: nullEntry}
+
+	matches := Route(context.Background(), snap, ev, now, lister)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (NoncurrentDays scheduled, NOT fired), got %v", matches)
+	}
+	// Successor = null mtime (today). Days threshold = 30. DueTime ≈ now+30d.
+	if !matches[0].DueTime.After(now.Add(29 * 24 * time.Hour)) {
+		t.Fatalf("DueTime=%v, want ~30d from null PUT (not from displaced version's age)", matches[0].DueTime)
+	}
+	if matches[0].VersionID != "v-old" {
+		t.Fatalf("VersionID=%q, want displaced v-old", matches[0].VersionID)
+	}
+}
+
+func TestRoutePointerTransitionSuspendedClearsPointerExpansionLatestPosIsNull(t *testing.T) {
+	// Same shape under a NewerNoncurrentVersions rule — exercises the
+	// expansion path. The new latest is "null", so latestPos must
+	// resolve to the null sibling. With a substitution latestIDForExpand
+	// = "null", the existing match-on-id logic finds it.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	staleMtime := now.AddDate(0, 0, -10).Unix()
+	ev := versionsContainerEventStaleMtime("bk", "logs/foo", "v-cur", staleMtime)
+
+	// Versions in .versions/, plus null. Mtime order newest-first:
+	// null (today), v-cur (-1d), v-mid (-2d), v-old (-30d).
+	// Post-flip ranks: latest=null, rank0=v-cur, rank1=v-mid, rank2=v-old.
+	// rank-2 v-old is the threshold-crosser. v-cur (rank0) and v-mid
+	// (rank1) retained.
+	listVersions := []*filer_pb.Entry{
+		displacedVersionEntry("v-cur", now.AddDate(0, 0, -1).Unix()),
+		displacedVersionEntry("v-mid", now.AddDate(0, 0, -2).Unix()),
+		displacedVersionEntry("v-old", now.AddDate(0, 0, -30).Unix()),
+	}
+	nullEntry := &filer_pb.Entry{
+		Name:       "foo",
+		Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix()},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte("null")},
+	}
+	lister := &recordingLister{listVersions: listVersions, nullEntry: nullEntry}
+
+	matches := Route(context.Background(), snap, ev, now, lister)
+	versionIDs := []string{}
+	for _, m := range matches {
+		versionIDs = append(versionIDs, m.VersionID)
+	}
+	if !contains(versionIDs, "v-old") {
+		t.Fatalf("rank-2 v-old must fire, matches=%v", versionIDs)
+	}
+	for _, id := range []string{"null", "v-cur", "v-mid"} {
+		if contains(versionIDs, id) {
+			t.Fatalf("%s must not fire (latest or retained), matches=%v", id, versionIDs)
+		}
+	}
+}

weed/s3api/s3lifecycle/scheduler/bootstrap.go

@@ -6,6 +6,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -28,14 +29,21 @@ type EventInjector interface {
 // 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. var so tests can shrink it
-// without producing thousands of entries.
-var listPageSize uint32 = 1024
+// 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
@@ -46,10 +54,10 @@ func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string
 				lastName = e.Name
 			}
 			return fn(e)
-		}, startFrom, false, listPageSize); err != nil {
+		}, startFrom, false, pageSize); err != nil {
 			return err
 		}
-		if pageCount < listPageSize {
+		if pageCount < pageSize {
 			return nil
 		}
 		startFrom = lastName

weed/s3api/s3lifecycle/scheduler/bootstrap_test.go

@@ -1009,9 +1009,9 @@ func TestExpandVersionsDir_PaginatesBeyondListingLimit(t *testing.T) {
 	// would silently truncate, so the rank/sort math would be wrong
 	// past the boundary. listAll paginates via StartFromFileName.
 	// Shrink listPageSize so the test doesn't need thousands of entries.
-	prevPageSize := listPageSize
-	listPageSize = 2
-	t.Cleanup(func() { listPageSize = prevPageSize })
+	prevPageSize := listPageSize.Load()
+	listPageSize.Store(2)
+	t.Cleanup(func() { listPageSize.Store(prevPageSize) })
 
 	now := time.Now().Truncate(time.Second)
 	const total = 7
@@ -1065,9 +1065,9 @@ func TestExpandVersionsDir_PaginatesBeyondListingLimit(t *testing.T) {
 func TestWalkBucketDir_PaginatesBeyondListingLimit(t *testing.T) {
 	// Same correctness story for the bucket-level walk: hot buckets
 	// with thousands of objects must not silently truncate.
-	prevPageSize := listPageSize
-	listPageSize = 2
-	t.Cleanup(func() { listPageSize = prevPageSize })
+	prevPageSize := listPageSize.Load()
+	listPageSize.Store(2)
+	t.Cleanup(func() { listPageSize.Store(prevPageSize) })
 
 	const total = 5
 	rootChildren := make([]*filer_pb.Entry, 0, total)