feat(s3/lifecycle): NoncurrentVersionExpiration via bootstrap (Phase 5b/3) (#9383)

* feat(s3/lifecycle): NoncurrentVersionExpiration via bootstrap (Phase 5b/3) Bootstrap now expands every <key>.versions/ directory into one event per version with sibling state pre-computed. The router fires NoncurrentDays / NewerNoncurrent off these events using SuccessorModTime as the noncurrent clock; previously these rules never ran on a versioned bucket because buildObjectInfo couldn't classify version-folder events without the latest pointer. Mechanics walkBucketDir treats a directory ending in .versions and carrying ExtLatestVersionIdKey as a SeaweedFS .versions container — emit it once and skip the recursion. Coincidentally-named directories without the latest pointer recurse normally. BucketBootstrapper.expandVersionsDir lists the children, sorts newest-first by mtime, resolves the latest position from the pointer, and injects a synthesized reader.Event per version with BootstrapVersion populated. NoncurrentIndex is 0-based among noncurrents in newest-first order; SuccessorModTime is the immediate newer sibling's mtime (zero for the latest). Pointer naming a missing or absent version falls back to the newest-by-mtime sibling so a race window can't flag every entry as noncurrent. routeBootstrapVersion uses BootstrapVersion to build ObjectInfo directly (bypassing the version-folder skip in buildObjectInfo) and runs the standard match loop. ABORT_MPU is excluded by kind-shape gate. The schedule clock uses SuccessorModTime for noncurrents and ModTime for the latest, so the dispatcher fires when the rule's days threshold is met. Match.ObjectKey is the LOGICAL key, Match.VersionID is the marker's stored version_id — the dispatcher reaches deleteSpecificObjectVersion or createDeleteMarker correctly. Layer 2 tests cover both sides. Router: latest fires ExpirationDays; noncurrent fires NoncurrentDays; NewerNoncurrentVersions retains the N newest noncurrents; ABORT_MPU never matches. Bootstrap: .versions dir emitted once and not recursed; missing latest pointer falls back to newest; backdated PUT (latest pointer is older by mtime) keeps the right noncurrent index; delete-marker flag propagates. * fix(s3/lifecycle): no VersionID for latest expirations, child-based dir disambig Two correctness gaps in Phase 5b/3. Bootstrap was pinning the version_id on every Match. For EXPIRATION_DAYS / EXPIRATION_DATE on the latest version this is unsafe: between schedule and dispatch a fresh PUT can land, the dispatcher would still identity-match against the original version's bytes (it still exists at that path) and the resulting delete marker would hide the new latest. Drop VersionID for those kinds; an 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. NoncurrentDays / NewerNoncurrent / EXPIRED_DELETE_MARKER still pin the version_id since those are version-targeted. isVersionsDir gating on ExtLatestVersionIdKey lost a race window: createDeleteMarker writes the version file before updating the parent's Extended pointer, so a walk between those two steps would see a .versions/ dir without the pointer, recurse into it, and emit raw version files that the router drops. Match the suffix only and let expandVersionsDir disambiguate by child inspection: if any child carries ExtVersionIdKey it's a real .versions container and we expand; otherwise it's a coincidentally-named user folder and we recurse via the bucket-walk's own callback so nested entries still flow through. Tests: latest-expiration assertion flipped to expect empty VersionID; new tests cover the coincidentally-named-folder recursion and the race-window expansion (children present, pointer absent). * fix(s3/lifecycle): filter directory + missing-version-id children at listing expandVersionsDir's listing callback collected every child with attributes; subdirectories or entries without ExtVersionIdKey would make it past the empty-id skip in the inner loop but still inflate NumVersions and skew NoncurrentIndex (the rank derives from the filtered slice's position, which was wrong when the unfiltered slice was sorted). Drop directories at listing time and partition the file children into a versions slice that's the actual rank source. Test cleanups: out-of-order-mtime test now sets v1 older than v2 so latestPos > 0 actually exercises the rank-skip branch in expandVersionsDir; bootstrapVersionEntry preserves nanosecond precision via MtimeNs to match markerLoneEntry's pattern; drop a leftover unused idx variable. * fix(s3/lifecycle): null version + canonical version-id tiebreak Two correctness gaps in Phase 5b/3 bootstrap. Null versions live at the bare logical path, not under .versions/. Bootstrap previously expanded only .versions/<key>/ children, so: - pre-versioning objects with newer .versions/ history never had their null version expired by NoncurrentDays - suspended-bucket writes (which clear the .versions/ latest pointer so null becomes current) had every .versions/ child wrongly classified as latest by the buildObjectInfo fallback expandVersionsDir now looks up the bare key via NewFullPath + LookupEntry, accepts a regular file or an explicit S3 directory-key marker (Mime set), and folds it into the sibling set with VersionID="null". Latest resolution: pointer present + names a real id wins; pointer absent + null exists makes null latest; otherwise falls back to newest sibling. The walker's regular emission for the bare entry would otherwise duplicate, so walkBucketDir now does a two-pass walk per directory level — .versions/ first, then everything else with a per-walk skipBare set keyed by bucket-relative path that expandVersionsDir populates when it claims a bare null sibling. Sort tiebreak: PUTs only set second-level Mtime, so two versions written in the same second tied. The unstable secondary order let old-format version filenames sort oldest-first and corrupt NoncurrentIndex under NewerNoncurrentVersions retention. Add CompareVersionIds to s3lifecycle/version_time.go (mirrors the canonical comparator in s3api/s3api_version_id.go to avoid the import cycle) and use it as a secondary key after mtime equality. Tests: pre-versioning null-as-noncurrent, suspended null-as-current, directory-key marker as null version, end-to-end claim through walkBucketDir's two-pass ordering, and same-second tiebreak via canonical version-id ordering. fakeFilerClient grows a LookupDirectoryEntry implementation backed by the same in-memory tree. * fix(s3/lifecycle): only treat explicit-null bare entries as current The pointer-missing branch in expandVersionsDir made null latest as soon as a bare object was found. That's correct for suspended-bucket writes (s3api_object_handlers_put.go writes the bare entry with ExtVersionIdKey="null") but wrong for the pre-versioning race window: a brand-new version under .versions/<file> exists before the parent's ExtLatestVersionIdKey update lands, and a pre-versioning bare object has no version-id marker. Marking that older bare object latest hides the real new version and skips noncurrent expiration of the null until the next process restart/bootstrap. Distinguish the two: lookupNullVersion now returns whether the bare entry's Extended map carries ExtVersionIdKey="null" (the suspended write marker). expandVersionsDir's pointer-missing branch only promotes null to latest when explicit; otherwise it falls back to newest-sibling, which is safe for the race window since the new version's mtime is fresher than the bare object's. The existing suspended-null test now uses a new helper that adds the explicit marker. New regression test covers the race window: bare entry without the marker + a fresh .versions/<v1> file + missing parent pointer must keep v1 as latest and the null as noncurrent. * fix(s3/lifecycle): only the newest item can be the explicit-null latest The pointer-missing branch in expandVersionsDir scanned every item for an explicit null and promoted it to latest. After a suspended->enabled transition that's the wrong call: createVersion writes the version file before updating ExtLatestVersionIdKey, so a bootstrap that lands in the race window sees an older bare null with ExtVersionIdKey="null" plus a newer .versions/<v-new> child and no parent pointer. Promoting the null misclassifies v-new as noncurrent and skips both the new version's current-version expiration and the null's noncurrent scheduling until the next bootstrap. Constrain the explicit-null branch to items[0]: if the suspended-null write is genuinely current it'll be the newest by mtime AND tagged. Anything else falls through to the newest-sibling default. Adds a regression test for the suspended->re-enabled race. * fix(s3/lifecycle): paginate bootstrap directory listings SeaweedList(..., limit=0) is a single-page request: the filer caps limit=0 at DirListingLimit (1000 by default) and returns whatever fits in one round trip. expandVersionsDir and walkBucketDir both relied on that, so any directory bigger than the cap silently truncated. For noncurrent retention this is correctness, not just scale — a hot key with more versions than the cap had its rank/sort math computed off the first page only, NumVersions, NoncurrentIndex, SuccessorModTime, and the latest-fallback all wrong, with the older versions never scheduled until a future bootstrap. Add a listAll helper that drives pagination via StartFromFileName + inclusive=false, looping until a page returns fewer entries than the configured page size. Use it in both call sites. Page size is a var (listPageSize, default 1024) so tests can shrink it without generating thousands of entries. The fake filer client now mirrors the real semantics: sort children by name, honor StartFromFileName/InclusiveStartFrom, cap at Limit. New regression tests force a small page size and assert the full result set is processed and the call count matches what pagination should drive. * perf(s3/lifecycle): stream bucket walk in two passes instead of buffering walkBucketDir was paginating into a children slice and then iterating twice (pass 1: .versions/, pass 2: everything else). For flat buckets with millions of entries the buffer is a real memory spike. Drop the materialization: each pass now drives its own listAll over the same directory and acts on entries as they stream in. The skipBare ordering contract is preserved — pass 2 still runs after pass 1 finishes — and the per-pass paging keeps memory bounded by listPageSize. Tradeoff: each directory level is listed twice. For workloads where that matters more than the memory headroom, we can revisit; the correctness/scale dial here is what the noncurrent rules need. Updated three tests for the new call count: each walk now records 2 listings per directory (pass 1 + pass 2). The KickOffNew dedup tests expect 2 calls per bucket; the pagination test expects 6 instead of 3.
2026-08-21 22:56:55 +00:00 · 2026-05-09 10:48:32 -07:00
parent 1c917ffacb
commit 2f7ac1d664
6 changed files with 1339 additions and 54 deletions
@@ -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.
@@ -71,6 +71,13 @@ func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now tim
 	if len(keys) == 0 {
 		return nil
 	}
+	// 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)

 	// EXP_DM can fire on two version-folder events: the marker create
@@ -220,6 +227,91 @@ func routeSoleSurvivorMarker(ctx context.Context, snap *engine.Snapshot, ev *rea
 	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.
@@ -828,3 +828,167 @@ func TestRouteVersionedAllVersionFolderPathsSkipped(t *testing.T) {
 	}
 }

+
+func bootstrapVersionEntry(versionID string, mtime time.Time, isDeleteMarker bool) *filer_pb.Entry {
+	ext := map[string][]byte{
+		s3_constants.ExtVersionIdKey: []byte(versionID),
+	}
+	if isDeleteMarker {
+		ext[s3_constants.ExtDeleteMarkerKey] = []byte("true")
+	}
+	mtimeUnix := mtime.Unix()
+	return &filer_pb.Entry{
+		Name: "v_" + versionID,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:   mtimeUnix,
+			MtimeNs: int32(mtime.UnixNano() - mtimeUnix*int64(1e9)),
+		},
+		Extended: ext,
+	}
+}
+
+func TestRouteBootstrapVersionLatestExpirationDaysFires(t *testing.T) {
+	// Bootstrap-emitted event for the LATEST version of a versioned
+	// object. ExpirationDays should fire (creates a delete marker at
+	// dispatch). ObjectKey is the LOGICAL key. VersionID must be EMPTY
+	// for EXPIRATION_DAYS so the dispatcher fetches the current latest:
+	// if a fresh PUT landed between schedule and dispatch, identity-CAS
+	// against the original version's bytes would pass even though the
+	// latest has moved on.
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -2)
+	entry := bootstrapVersionEntry("v-current", old, false)
+	ev := &reader.Event{
+		Bucket:   "bk",
+		Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_v-current",
+		NewEntry: entry,
+		BootstrapVersion: &reader.BootstrapVersion{
+			LogicalKey:  "logs/foo",
+			VersionID:   "v-current",
+			IsLatest:    true,
+			NumVersions: 1,
+		},
+	}
+	matches := Route(context.Background(), snap, ev, now, nil)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (ExpirationDays on latest), got %v", matches)
+	}
+	if matches[0].ObjectKey != "logs/foo" {
+		t.Fatalf("ObjectKey=%q, want logs/foo", matches[0].ObjectKey)
+	}
+	if matches[0].VersionID != "" {
+		t.Fatalf("VersionID=%q, want empty for EXPIRATION_DAYS", matches[0].VersionID)
+	}
+}
+
+func TestRouteBootstrapVersionNoncurrentDaysFires(t *testing.T) {
+	// Bootstrap-emitted event for a NONCURRENT version. NoncurrentDays
+	// uses SuccessorModTime as the clock — when this version was
+	// replaced by the next-newer sibling.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	successor := now.AddDate(0, 0, -3) // replaced 3 days ago
+	old := now.AddDate(0, 0, -10)      // mtime older still
+	entry := bootstrapVersionEntry("v-old", old, false)
+	ev := &reader.Event{
+		Bucket:   "bk",
+		Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_v-old",
+		NewEntry: entry,
+		BootstrapVersion: &reader.BootstrapVersion{
+			LogicalKey:       "logs/foo",
+			VersionID:        "v-old",
+			IsLatest:         false,
+			NumVersions:      2,
+			NoncurrentIndex:  0,
+			SuccessorModTime: successor,
+		},
+	}
+	matches := Route(context.Background(), snap, ev, now, nil)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (NoncurrentDays), got %v", matches)
+	}
+	if matches[0].VersionID != "v-old" {
+		t.Fatalf("VersionID=%q, want v-old", matches[0].VersionID)
+	}
+}
+
+func TestRouteBootstrapVersionNoncurrentRespectsNewerNoncurrentVersions(t *testing.T) {
+	// NewerNoncurrentVersions=2 keeps the two newest noncurrents safe.
+	// A version at NoncurrentIndex=0 (newest noncurrent) must NOT fire;
+	// index=2 (third-newest) MUST fire.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	successor := now.AddDate(0, 0, -3)
+	old := now.AddDate(0, 0, -10)
+
+	mk := func(idx int) *reader.Event {
+		return &reader.Event{
+			Bucket:   "bk",
+			Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_old" + string(rune('0'+idx)),
+			NewEntry: bootstrapVersionEntry("old"+string(rune('0'+idx)), old, false),
+			BootstrapVersion: &reader.BootstrapVersion{
+				LogicalKey:       "logs/foo",
+				VersionID:        "old" + string(rune('0'+idx)),
+				IsLatest:         false,
+				NumVersions:      4,
+				NoncurrentIndex:  idx,
+				SuccessorModTime: successor,
+			},
+		}
+	}
+
+	if got := Route(context.Background(), snap, mk(0), now, nil); len(got) != 0 {
+		t.Fatalf("noncurrent rank 0 must be retained, got %v", got)
+	}
+	if got := Route(context.Background(), snap, mk(1), now, nil); len(got) != 0 {
+		t.Fatalf("noncurrent rank 1 must be retained, got %v", got)
+	}
+	if got := Route(context.Background(), snap, mk(2), now, nil); len(got) != 1 {
+		t.Fatalf("noncurrent rank 2 must fire, got %v", got)
+	}
+}
+
+func TestRouteBootstrapVersionAbortMPUNeverEmittedForVersion(t *testing.T) {
+	// AbortIncompleteMultipartUpload only applies to MPU init dirs, not
+	// versioned object versions. Even if the bucket has the rule, a
+	// bootstrap version event must not produce an ABORT_MPU match.
+	rule := &s3lifecycle.Rule{
+		ID:                          "r",
+		Status:                      s3lifecycle.StatusEnabled,
+		AbortMPUDaysAfterInitiation: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -10)
+	ev := &reader.Event{
+		Bucket:   "bk",
+		Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_x",
+		NewEntry: bootstrapVersionEntry("x", old, false),
+		BootstrapVersion: &reader.BootstrapVersion{
+			LogicalKey:  "logs/foo",
+			VersionID:   "x",
+			IsLatest:    true,
+			NumVersions: 1,
+		},
+	}
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
+		t.Fatalf("bootstrap version event must not produce ABORT_MPU match, got %v", got)
+	}
+}
@@ -3,14 +3,17 @@ package scheduler
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
+	"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 +23,39 @@ 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. var so tests can shrink it
+// without producing thousands of entries.
+var listPageSize uint32 = 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 {
+	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, listPageSize); err != nil {
+			return err
+		}
+		if pageCount < listPageSize {
+			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,
@@ -75,7 +111,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,7 +143,8 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 		}
 		count++
 		return b.Injector.InjectEvent(ctx, ev)
-	}); err != nil {
+	}
+	if err := walkBucketDir(ctx, b.FilerClient, root, root, cb); err != nil {
 		if ctx.Err() == nil {
 			glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, err)
 		}
@@ -97,42 +153,244 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 	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 +410,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)
+}
+
@@ -3,7 +3,9 @@ package scheduler
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -52,8 +54,8 @@ func (s *fakeListStream) SendMsg(any) error { return nil }
 func (s *fakeListStream) RecvMsg(any) error { return nil }

 // fakeFilerClient embeds SeaweedFilerClient (nil interface) and overrides
-// only ListEntries. Calling any other method panics, which is fine for
-// these tests.
+// ListEntries + LookupDirectoryEntry. Calling any other method panics,
+// which is fine for these tests.
 type fakeFilerClient struct {
 	filer_pb.SeaweedFilerClient

@@ -63,15 +65,49 @@ type fakeFilerClient struct {
 	listedN  int32                        // atomic counter for cross-goroutine reads
 }

+func (c *fakeFilerClient) LookupDirectoryEntry(ctx context.Context, in *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, e := range c.tree[in.Directory] {
+		if e != nil && e.Name == in.Name {
+			return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil
+		}
+	}
+	return nil, filer_pb.ErrNotFound
+}
+
 func (c *fakeFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
 	c.mu.Lock()
 	c.listed = append(c.listed, in.Directory)
 	c.mu.Unlock()
 	atomic.AddInt32(&c.listedN, 1)

-	children := c.tree[in.Directory]
-	responses := make([]*filer_pb.ListEntriesResponse, 0, len(children))
-	for _, e := range children {
+	// Mirror the filer: sort children by name, honor StartFromFileName
+	// (exclusive unless InclusiveStartFrom), and cap at Limit. listAll's
+	// pagination loop depends on these semantics to advance correctly.
+	src := c.tree[in.Directory]
+	filtered := make([]*filer_pb.Entry, 0, len(src))
+	for _, e := range src {
+		if e == nil {
+			continue
+		}
+		if in.StartFromFileName != "" {
+			if in.InclusiveStartFrom {
+				if e.Name < in.StartFromFileName {
+					continue
+				}
+			} else if e.Name <= in.StartFromFileName {
+				continue
+			}
+		}
+		filtered = append(filtered, e)
+	}
+	sort.SliceStable(filtered, func(i, j int) bool { return filtered[i].Name < filtered[j].Name })
+	if in.Limit > 0 && uint32(len(filtered)) > in.Limit {
+		filtered = filtered[:in.Limit]
+	}
+	responses := make([]*filer_pb.ListEntriesResponse, 0, len(filtered))
+	for _, e := range filtered {
 		responses = append(responses, &filer_pb.ListEntriesResponse{Entry: e})
 	}
 	return &fakeListStream{responses: responses, ctx: ctx}, nil
@@ -345,13 +381,18 @@ func TestBucketBootstrapper_KickOffNew_LaunchesPerBucket(t *testing.T) {

 	b.KickOffNew(context.Background(), []string{"bucketA", "bucketB"})

-	// Both walks must hit ListEntries once each (empty trees -> no recursion).
+	// Each walk lists the bucket root twice (pass 1: .versions/, pass 2:
+	// everything else); 2 buckets * 2 passes = 4 listings total.
 	waitFor(t, func() bool {
-		return atomic.LoadInt32(&client.listedN) >= 2
-	}, "both bucket walks to start")
+		return atomic.LoadInt32(&client.listedN) >= 4
+	}, "both bucket walks to complete")

 	listed := client.listedCopy()
-	assert.ElementsMatch(t, []string{"/buckets/bucketA", "/buckets/bucketB"}, listed)
+	seen := map[string]bool{}
+	for _, d := range listed {
+		seen[d] = true
+	}
+	assert.Equal(t, map[string]bool{"/buckets/bucketA": true, "/buckets/bucketB": true}, seen)

 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -370,31 +411,34 @@ func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) {
 	}

 	b.KickOffNew(context.Background(), []string{"bucketA", "bucketB"})
+	// Each walk does two ListEntries calls (pass 1: .versions/, pass 2:
+	// everything else). 2 buckets * 2 passes = 4 listings.
 	waitFor(t, func() bool {
-		return atomic.LoadInt32(&client.listedN) >= 2
+		return atomic.LoadInt32(&client.listedN) >= 4
 	}, "first wave to complete")

 	firstWave := atomic.LoadInt32(&client.listedN)

 	// Second call: bucketA is already known, bucketC is new. Only one
-	// new walk should fire.
+	// new walk should fire (2 listings).
 	b.KickOffNew(context.Background(), []string{"bucketA", "bucketC"})
 	waitFor(t, func() bool {
-		return atomic.LoadInt32(&client.listedN) >= firstWave+1
-	}, "bucketC walk to start")
+		return atomic.LoadInt32(&client.listedN) >= firstWave+2
+	}, "bucketC walk to complete")

 	// Give a moment for any spurious bucketA walk to also tick.
 	time.Sleep(20 * time.Millisecond)

 	listed := client.listedCopy()
-	// Count distinct buckets walked.
+	// Each bucket walks once across both calls; the walk does two
+	// listings of the bucket root (pass 1 + pass 2).
 	bucketCount := map[string]int{}
 	for _, d := range listed {
 		bucketCount[d]++
 	}
-	assert.Equal(t, 1, bucketCount["/buckets/bucketA"], "bucketA must be walked exactly once across both calls")
-	assert.Equal(t, 1, bucketCount["/buckets/bucketB"])
-	assert.Equal(t, 1, bucketCount["/buckets/bucketC"])
+	assert.Equal(t, 2, bucketCount["/buckets/bucketA"], "bucketA must be walked exactly once (=2 list calls) across both calls")
+	assert.Equal(t, 2, bucketCount["/buckets/bucketB"])
+	assert.Equal(t, 2, bucketCount["/buckets/bucketC"])

 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -435,3 +479,624 @@ func TestBucketBootstrapper_KickOffNew_EmptyBucketListIsNoop(t *testing.T) {
 	assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN))
 	assert.Empty(t, inj.snapshot())
 }
+
+// versionFile builds a version-file entry with the given mtime, version_id,
+// and optional delete-marker flag.
+func versionFile(versionID string, mtime time.Time, isMarker bool) *filer_pb.Entry {
+	ext := map[string][]byte{
+		s3_constants.ExtVersionIdKey: []byte(versionID),
+	}
+	if isMarker {
+		ext[s3_constants.ExtDeleteMarkerKey] = []byte("true")
+	}
+	return &filer_pb.Entry{
+		Name: "v_" + versionID,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: mtime.Unix(),
+		},
+		Extended: ext,
+	}
+}
+
+func TestWalkBucketDir_VersionsDirEmittedOnceAndNotRecursed(t *testing.T) {
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v2"),
+	})
+	now := time.Now()
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", now.Add(-2*time.Hour), false),
+				versionFile("v2", now, false),
+			},
+		},
+	}
+	var seen []string
+	err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"foo" + s3_constants.VersionsFolder}, seen)
+	assert.NotContains(t, client.listedCopy(), testBucketRoot+"/foo"+s3_constants.VersionsFolder)
+}
+
+func TestWalkBucketDir_VersionsDirEmittedRegardlessOfLatestPointer(t *testing.T) {
+	// walkBucketDir matches <x>.versions/ purely on the name suffix —
+	// gating on ExtLatestVersionIdKey would lose the race window where
+	// the version file exists before the parent's metadata update lands.
+	// expandVersionsDir handles disambiguation by inspecting children.
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, nil)
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {fileEntry("inner.txt")},
+		},
+	}
+	var seen []string
+	err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"foo" + s3_constants.VersionsFolder}, seen)
+}
+
+func TestExpandVersionsDir_CoincidentallyNamedFolderRecursesViaFallback(t *testing.T) {
+	// A user-created folder happening to end in .versions/ has children
+	// without ExtVersionIdKey. expandVersionsDir must recurse via the
+	// fallback callback so inner files still emit normal events.
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, nil)
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				fileEntry("inner.txt"),
+				dirEntry("sub", nil),
+			},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder + "/sub": {fileEntry("deep.txt")},
+		},
+	}
+	var seen []string
+	cb := func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets"}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, cb, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 0, count, "fallback path returns 0 — events go through cb")
+	assert.ElementsMatch(t, []string{"foo" + s3_constants.VersionsFolder + "/inner.txt", "foo" + s3_constants.VersionsFolder + "/sub/deep.txt"}, seen)
+}
+
+func TestExpandVersionsDir_RaceWithMissingPointerStillExpands(t *testing.T) {
+	// Real .versions container whose parent metadata update hasn't
+	// landed yet (no ExtLatestVersionIdKey on the dir). Children DO
+	// carry ExtVersionIdKey. expandVersionsDir must still emit version
+	// events; missing-pointer fallback (newest-by-mtime as latest)
+	// covers retention safety.
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, nil) // no Extended at all
+	now := time.Now()
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", now.Add(-2*time.Hour), false),
+				versionFile("v2", now.Add(-1*time.Hour), false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 2, count)
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v2"].IsLatest, "newest by mtime is latest when pointer missing")
+}
+
+func TestExpandVersionsDir_LatestAndNoncurrentsByMtime(t *testing.T) {
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-2 * time.Hour)
+	v3mt := now.Add(-1 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v3"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+				versionFile("v2", v2mt, false),
+				versionFile("v3", v3mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 3, count)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		require.NotNil(t, ev.BootstrapVersion)
+		assert.Equal(t, "foo", ev.BootstrapVersion.LogicalKey)
+		assert.Equal(t, 3, ev.BootstrapVersion.NumVersions)
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Contains(t, byID, "v1")
+	require.Contains(t, byID, "v2")
+	require.Contains(t, byID, "v3")
+
+	assert.True(t, byID["v3"].IsLatest)
+	assert.True(t, byID["v3"].SuccessorModTime.IsZero(), "newest sibling has no successor")
+
+	assert.False(t, byID["v2"].IsLatest)
+	assert.Equal(t, 0, byID["v2"].NoncurrentIndex, "newest noncurrent")
+	assert.Equal(t, v3mt.Unix(), byID["v2"].SuccessorModTime.Unix())
+
+	assert.False(t, byID["v1"].IsLatest)
+	assert.Equal(t, 1, byID["v1"].NoncurrentIndex)
+	assert.Equal(t, v2mt.Unix(), byID["v1"].SuccessorModTime.Unix())
+}
+
+func TestExpandVersionsDir_LatestPointerOutOfOrderByMtime(t *testing.T) {
+	// Backdated PUT scenario: latest pointer names v1 but v1's mtime is
+	// OLDER than v2's. After newest-first sort the order is [v2, v1] so
+	// latestPos == 1, exercising the rank-skip path for the noncurrent.
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-1 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+				versionFile("v2", v2mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v1"].IsLatest)
+	assert.False(t, byID["v2"].IsLatest)
+	assert.Equal(t, 0, byID["v2"].NoncurrentIndex, "v2 is the only noncurrent → rank 0")
+}
+
+func TestExpandVersionsDir_MissingLatestPointerFallsBackToNewest(t *testing.T) {
+	// No latest pointer (rare race window): treat the newest sibling
+	// by mtime as latest so retention isn't unsafe.
+	now := time.Now()
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", now.Add(-2*time.Hour), false),
+				versionFile("v2", now.Add(-1*time.Hour), false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v2"].IsLatest, "newest by mtime is latest when pointer missing")
+	assert.False(t, byID["v1"].IsLatest)
+}
+
+func TestExpandVersionsDir_DeleteMarkerFlagPropagated(t *testing.T) {
+	now := time.Now()
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v-marker"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v-marker", now, true),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+
+	events := inj.snapshot()
+	require.Len(t, events, 1)
+	assert.True(t, events[0].BootstrapVersion.IsDeleteMarker)
+	assert.True(t, events[0].BootstrapVersion.IsLatest)
+}
+
+func bareFile(name string, mtime time.Time) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: name,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: mtime.Unix(),
+		},
+	}
+}
+
+// suspendedNullFile mirrors the suspended-versioning write path: the
+// bare entry carries ExtVersionIdKey="null" so bootstrap can tell it
+// apart from a pre-versioning bare object during a pointer-missing
+// race window.
+func suspendedNullFile(name string, mtime time.Time) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: name,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: mtime.Unix(),
+		},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte("null"),
+		},
+	}
+}
+
+func TestExpandVersionsDir_PreVersioningNullIsNoncurrent(t *testing.T) {
+	// Object existed pre-versioning as the bare key. Versioning was
+	// enabled and a newer version v1 was PUT under .versions/. The
+	// .versions/ latest pointer names v1, so null is noncurrent.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour)
+	nullMt := now.Add(-3 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {bareFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+	assert.Equal(t, 2, count)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v1"].IsLatest)
+	assert.False(t, byID["null"].IsLatest)
+	assert.Equal(t, 0, byID["null"].NoncurrentIndex)
+	assert.Equal(t, 2, byID["null"].NumVersions)
+	assert.True(t, skipBare["foo"], "bare-key skip recorded")
+}
+
+func TestExpandVersionsDir_SuspendedNullIsCurrent(t *testing.T) {
+	// Suspended-bucket scenario: a write to the null version cleared the
+	// .versions/ latest pointer AND tagged the bare entry with
+	// ExtVersionIdKey="null". Older real versions remain in .versions/.
+	// Null must be IsLatest=true; .versions/ children become noncurrent.
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-2 * time.Hour)
+	nullMt := now.Add(-1 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{}) // pointer cleared
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {suspendedNullFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+				versionFile("v2", v2mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["null"].IsLatest, "pointer cleared + null exists -> null is latest")
+	assert.False(t, byID["v1"].IsLatest)
+	assert.False(t, byID["v2"].IsLatest)
+	assert.True(t, skipBare["foo"])
+}
+
+func TestExpandVersionsDir_NullVersionDirectoryKeyMarker(t *testing.T) {
+	// Directory-key marker (object name ends in /): the bare entry is a
+	// directory with Mime set. Treat as null version.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour)
+	dirMarker := &filer_pb.Entry{
+		Name:        "foo",
+		IsDirectory: true,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: now.Add(-3 * time.Hour).Unix(),
+			Mime:  "application/x-directory",
+		},
+	}
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {dirMarker, versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Contains(t, byID, "null", "directory-key marker counts as null version")
+	assert.True(t, skipBare["foo"])
+}
+
+func TestWalkBucketDir_VersionsDirOrderingClaimsNullSibling(t *testing.T) {
+	// End-to-end through walkBucket: bare foo + foo.versions/ are
+	// siblings in the same directory. The two-pass walker processes
+	// the .versions/ first; expandVersionsDir claims "foo" as null;
+	// the second pass sees skipBare["foo"]==true and emits no regular
+	// event for the bare entry.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour)
+	nullMt := now.Add(-3 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {bareFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	b.KickOffNew(context.Background(), []string{"b1"})
+	// give the goroutine time to finish
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) && len(inj.snapshot()) < 2 {
+		time.Sleep(10 * time.Millisecond)
+	}
+	events := inj.snapshot()
+
+	// Exactly two events: v1 (latest) and null (noncurrent). NO third
+	// regular event for the bare "foo" entry.
+	assert.Len(t, events, 2)
+	versionIDs := []string{}
+	for _, ev := range events {
+		require.NotNil(t, ev.BootstrapVersion, "all events must be BootstrapVersion-tagged")
+		versionIDs = append(versionIDs, ev.BootstrapVersion.VersionID)
+	}
+	assert.ElementsMatch(t, []string{"v1", "null"}, versionIDs)
+}
+
+func TestExpandVersionsDir_VersionIDTiebreakOnSameSecondMtime(t *testing.T) {
+	// Two versions written in the same second: Mtime ties. The
+	// CompareVersionIds tiebreak puts the version_id with newer
+	// canonical ordering first. Use new-format IDs (inverted timestamps)
+	// so smaller string sorts as newer.
+	now := time.Now().Truncate(time.Second)
+	idNewer := "8000000000000000aaaaaaaaaaaaaaaa"
+	idOlder := "9000000000000000bbbbbbbbbbbbbbbb"
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte(idNewer),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile(idOlder, now, false),
+				versionFile(idNewer, now, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID[idNewer].IsLatest, "newer canonical id wins the tiebreak")
+	assert.False(t, byID[idOlder].IsLatest)
+	assert.Equal(t, 0, byID[idOlder].NoncurrentIndex)
+}
+
+func TestExpandVersionsDir_PreVersioningNullDuringPointerRaceFallsBackToNewest(t *testing.T) {
+	// Pre-versioning bare object existed when versioning was enabled.
+	// A new version v1 was just written under .versions/<file> but the
+	// parent's ExtLatestVersionIdKey update has not landed yet. The
+	// bare entry has NO ExtVersionIdKey marker — distinguishing it from
+	// a suspended-bucket write. Bootstrap must treat v1 as latest (the
+	// newest sibling) and the implicit null as noncurrent, so the null
+	// expiration is scheduled this run instead of waiting for a future
+	// bootstrap.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour) // newer
+	nullMt := now.Add(-3 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{}) // pointer not yet written
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {bareFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v1"].IsLatest, "newest sibling wins when null is implicit")
+	assert.False(t, byID["null"].IsLatest, "implicit null is noncurrent during pointer-missing race")
+}
+
+func TestExpandVersionsDir_SuspendedThenReEnabledNullIsNoncurrent(t *testing.T) {
+	// Bucket was suspended: bare entry was written with
+	// ExtVersionIdKey="null" and the .versions/ pointer cleared.
+	// Versioning was re-enabled and a fresh PUT created
+	// .versions/<v-new> with newer mtime, but the pointer-update for
+	// that new version hasn't landed yet. Bootstrap running in this
+	// window must keep v-new as latest (it's newest by mtime); the
+	// explicit null is noncurrent. Promoting the older null to latest
+	// just because it's explicit would skip current-version expiration
+	// of v-new and never schedule the null's noncurrent retention.
+	now := time.Now()
+	nullMt := now.Add(-3 * time.Hour) // OLDER bare-null
+	vNewMt := now.Add(-1 * time.Hour) // newer real version
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{}) // pointer not yet written
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {suspendedNullFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v-new", vNewMt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v-new"].IsLatest, "newest sibling wins even when an older explicit null exists")
+	assert.False(t, byID["null"].IsLatest)
+	assert.Equal(t, 0, byID["null"].NoncurrentIndex)
+}
+
+func TestExpandVersionsDir_PaginatesBeyondListingLimit(t *testing.T) {
+	// The filer caps SeaweedList(..., limit=0) at DirListingLimit per
+	// call. Expanding a hot key with more versions than that limit
+	// 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 })
+
+	now := time.Now().Truncate(time.Second)
+	const total = 7
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v0"),
+	})
+	versions := make([]*filer_pb.Entry, 0, total)
+	for i := 0; i < total; i++ {
+		// v0 is newest; v6 is oldest. Names are v_v00 .. v_v06 so the
+		// sort-by-name in the fake matches the sort-by-mtime here.
+		versions = append(versions, versionFile(fmt.Sprintf("v%02d", i), now.Add(-time.Duration(i)*time.Hour), false))
+	}
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: versions,
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, total, count, "every page must reach the injector")
+
+	listed := client.listedCopy()
+	calls := 0
+	for _, d := range listed {
+		if d == testBucketRoot+"/foo"+s3_constants.VersionsFolder {
+			calls++
+		}
+	}
+	// Pages of 2 over 7 items = 4 calls (2+2+2+1). Loop exits once
+	// page count < listPageSize on the 4th call.
+	assert.Equal(t, 4, calls, "must paginate via StartFromFileName")
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Len(t, byID, total, "no version dropped at a page boundary")
+	for i := 0; i < total; i++ {
+		bv := byID[fmt.Sprintf("v%02d", i)]
+		require.NotNil(t, bv)
+		assert.Equal(t, total, bv.NumVersions, "NumVersions reflects every page")
+	}
+	// v0 is the latest pointer target and the newest by mtime.
+	assert.True(t, byID["v00"].IsLatest)
+	// v6 is the oldest noncurrent.
+	assert.Equal(t, total-2, byID["v06"].NoncurrentIndex)
+}
+
+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 })
+
+	const total = 5
+	rootChildren := make([]*filer_pb.Entry, 0, total)
+	for i := 0; i < total; i++ {
+		rootChildren = append(rootChildren, fileEntry(fmt.Sprintf("k%02d", i)))
+	}
+	client := &fakeFilerClient{tree: map[string][]*filer_pb.Entry{testBucketRoot: rootChildren}}
+	var seen []string
+	err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	})
+	require.NoError(t, err)
+
+	want := make([]string, 0, total)
+	for i := 0; i < total; i++ {
+		want = append(want, fmt.Sprintf("k%02d", i))
+	}
+	assert.ElementsMatch(t, want, seen, "every page processed")
+
+	listed := client.listedCopy()
+	calls := 0
+	for _, d := range listed {
+		if d == testBucketRoot {
+			calls++
+		}
+	}
+	// Each directory is streamed twice (pass 1: .versions/, pass 2:
+	// everything else) to keep memory bounded on flat buckets. 5
+	// entries at page 2 = 3 paginated calls per pass = 6 total.
+	assert.Equal(t, 6, calls)
+}
@@ -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)
+}