feat(s3): stamp noncurrent_since on versioned demotions (#9431)

* feat(s3): stamp noncurrent_since on versioned demotions

A version's noncurrent TTL clock starts when the next version is
written, not at its own mtime. Today the lifecycle engine derives
that moment from the next-newer sibling's mtime — a heuristic that
drifts if the sibling is later modified and is unavailable when
the demoting event sits outside meta-log retention.

Stamp Seaweed-X-Amz-Noncurrent-Since-Ns on the demoted entry at
the two places where a PUT flips the latest pointer:
updateLatestVersionInDirectory and
updateIsLatestFlagsForSuspendedVersioning. Timestamp source is
time.Now().UnixNano() captured once per demotion — the documented
Phase 1 fallback until the filer write API surfaces its own TsNs.

Engine reads the stamp on both the bootstrap walker path and the
event-driven router; missing/zero falls back to the legacy
sibling-mtime derivation, so pre-stamp entries keep working.

Prerequisite for the daily-replay lifecycle worker (Phase 2+).

* fix(s3): address CI failure and PR review feedback

- Backdating tests must move both clocks: the lifecycle integration
  tests backdate version mtimes to simulate aging, but my earlier
  commit made the engine prefer the explicit demotion stamp over
  sibling mtime, so a real-now stamp dominated a backdated mtime and
  the rule never fired. Update backdateVersionedMtime to also rewrite
  Seaweed-X-Amz-Noncurrent-Since-Ns when the entry already carries it.
  This is a test simplification — production stamps record when the
  successor was written, not the demoted version's own mtime — but the
  resulting clock is correctly old enough.

- Refactor stamp parsing into one shared helper. Per gemini-code-assist:
  the parsing logic for ExtNoncurrentSinceNsKey was duplicated in
  router/router.go and scheduler/bootstrap.go. Move it to a new
  weed/s3api/s3lifecycle/noncurrent_since.go as exported
  SuccessorFromEntryStamp; both call sites now go through it.

- Make the parser ordering test deterministic. Per coderabbitai:
  time.Now().UnixNano() drops the monotonic clock component, so
  two back-to-back calls can decrease if the wall clock steps
  backward — the prior test was exercising OS clock behavior rather
  than the parser. Replace with fixed nanosecond values.

- Close a suspended-versioning race. Per coderabbitai: the prior
  putSuspendedVersioningObject called updateIsLatestFlagsForSuspendedVersioning
  after putToFiler returned, i.e. after the object write lock released.
  A concurrent PUT could promote a newer latest version, which we'd
  then wipe — leaving the older "null" object incorrectly current.
  Move the cleanup into the afterCreate callback so the null write and
  the .versions pointer clear (including the new demotion stamp) run
  atomically under the same lock. Best-effort logging is preserved.

* fix(s3/lifecycle): clear noncurrent_since stamp on test backdate

Backdating a version's mtime in tests is not a coherent claim about
when it became noncurrent — production stamps record the successor's
PUT time, which the test doesn't manipulate. The prior commit rewrote
the stamp to the backdated instant, but for TestLifecycleNewerNoncurrent
that creates an inconsistent state: v3's stamp says "demoted 30 days
ago" while v4's mtime (the supposed demoter) is real-now. With both
NewerNoncurrentVersions and NoncurrentDays in the same rule, the
NoncurrentDays floor passes against the backdated stamp and the
rank-based check then deletes v3 via the meta-log historical replay
that misranks against current state.

Clearing the stamp instead lets the lifecycle engine fall back to the
sibling-mtime derivation the tests were originally written against:
the legacy code path is preserved end-to-end while the new explicit-
stamp path is exercised by the unit tests in s3lifecycle/noncurrent_since_test.go
and the bootstrap-walker integration in scheduler/bootstrap_test.go.

The deeper interaction — historical meta-log replay ranking against
current state inside routePointerTransitionExpand — is pre-existing
and is no longer masked by the freshly-PUT successor's mtime once the
stamp is read. Tracked separately; not blocking this PR.

* fix(s3): stamp noncurrent_since before the .versions/ pointer flip

The pointer-flip on the .versions/ directory emits a meta-log event that
the lifecycle router consumes via routePointerTransition. The router
then calls LookupVersion on the demoted version's id. With the prior
ordering — pointer flip first, stamp second — the router could read
the demoted entry before markVersionNoncurrent landed and fall back to
the legacy sibling-mtime derivation.

Versioned COPY is the clean break: the new latest version keeps the
source object's mtime instead of recording the moment v_old was
demoted, so the fallback's successor clock can be arbitrarily wrong.
Reorder both updateLatestVersionInDirectory and
updateIsLatestFlagsForSuspendedVersioning so the stamp is written
first; the pointer flip then emits an event into a state where the
stamp is already present.

Failure of the stamp write remains non-fatal — lifecycle still falls
back to the legacy derivation in that case, with the same caveats as
before the PR but no race window.

test/s3/lifecycle/s3_lifecycle_versioning_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/stretchr/testify/require"
 )
 
@@ -54,6 +55,15 @@ func putNoncurrentExpirationLifecycle(t *testing.T, c *s3.Client, bucket, prefix
 
 // backdateVersionedMtime ages a specific .versions/v_<id> entry. The
 // version files live under <buckets>/<bucket>/<key>.versions/v_<versionId>.
+//
+// Also clears the ExtNoncurrentSinceNsKey stamp if present. The stamp
+// records when this version was demoted by its successor's PUT; tests
+// that backdate a version's own mtime aren't expressing a coherent claim
+// about the demotion moment, so leaving a real-now stamp would let it
+// dominate the backdated mtime and stop the rule from firing. Clearing
+// makes the lifecycle engine fall back to the sibling-mtime derivation,
+// which is what these tests were originally written against. Production
+// PUTs always write a stamp; this is a test-only escape hatch.
 func backdateVersionedMtime(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket, key, versionID string, daysOld int) {
 	t.Helper()
 	dir := bucketsPath + "/" + bucket + "/" + key + ".versions"
@@ -67,6 +77,7 @@ func backdateVersionedMtime(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket
 
 	resp.Entry.Attributes.Mtime = time.Now().Add(-time.Duration(daysOld) * 24 * time.Hour).Unix()
 	resp.Entry.Attributes.MtimeNs = 0
+	delete(resp.Entry.Extended, s3_constants.ExtNoncurrentSinceNsKey)
 	_, err = fc.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
 		Directory: dir, Entry: resp.Entry,
 	})

weed/s3api/s3_constants/extend_key.go

@@ -19,6 +19,12 @@ const (
 	ExtLatestVersionOwnerKey       = "Seaweed-X-Amz-Latest-Version-Owner"
 	ExtLatestVersionIsDeleteMarker = "Seaweed-X-Amz-Latest-Version-Is-Delete-Marker"
 	ExtMultipartObjectKey          = "key"
+	// Wall-clock nanoseconds (int64 as decimal string) captured at the
+	// moment a versioned entry was demoted from current to noncurrent
+	// by a later PUT or delete marker. Read by the s3 lifecycle engine
+	// to compute NoncurrentDays due time; zero/missing falls back to
+	// the entry's own mtime so legacy data still expires.
+	ExtNoncurrentSinceNsKey = "Seaweed-X-Amz-Noncurrent-Since-Ns"
 
 	// S3 checksum storage keys (use x-seaweedfs- prefix to avoid leaking in generic header loop)
 	ExtChecksumAlgorithm = "x-seaweedfs-checksum-algorithm"

weed/s3api/s3api_object_handlers_put.go

@@ -1274,19 +1274,25 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
 	// Upload the file using putToFiler - this will create the file with version metadata.
 	// Versioned/suspended bucket → resolver returns 0 by construction;
 	// pass 0 directly so the path is explicit at the call site.
-	etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, nil)
+	//
+	// Clear the prior latest-version pointer (and stamp the displaced
+	// entry with NoncurrentSinceNs) inside the afterCreate callback so
+	// it runs while withObjectWriteLock is still held in putToFiler.
+	// Doing it after putToFiler returns would race a concurrent PUT
+	// promoting a newer latest, which we'd then incorrectly wipe.
+	etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, func(_ *filer_pb.Entry) s3err.ErrorCode {
+		if err := s3a.updateIsLatestFlagsForSuspendedVersioning(bucket, normalizedObject); err != nil {
+			// Best-effort: a stale IsLatest flag is recoverable on the
+			// next list-versions resync, so don't fail the PUT.
+			glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
+		}
+		return s3err.ErrNone
+	})
 	if errCode != s3err.ErrNone {
 		glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
 		return "", errCode, SSEResponseMetadata{}
 	}
 
-	// Update all existing versions/delete markers to set IsLatest=false since "null" is now latest
-	err = s3a.updateIsLatestFlagsForSuspendedVersioning(bucket, normalizedObject)
-	if err != nil {
-		glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
-		// Don't fail the request, but log the warning
-	}
-
 	glog.V(2).Infof("putSuspendedVersioningObject: successfully created null version for %s/%s", bucket, object)
 
 	return etag, s3err.ErrNone, sseMetadata
@@ -1341,6 +1347,21 @@ 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 {
+		// Capture previously-latest filename before clearing the pointer.
+		prevLatestFileName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey])
+
+		// Stamp the demoted version BEFORE the .versions/ pointer
+		// clear. The pointer clear emits a meta-log event the lifecycle
+		// router consumes, then looks up the demoted version. If the
+		// stamp isn't already on the entry, the router falls back to a
+		// sibling-mtime derivation that can be wrong (versioned COPY
+		// in particular keeps the source's mtime). Best-effort:
+		// transient stamp-write failures are logged, lifecycle then
+		// falls back to the legacy derivation.
+		if prevLatestFileName != "" {
+			s3a.markVersionNoncurrent(bucketDir, versionsObjectPath, prevLatestFileName, time.Now().UnixNano())
+		}
+
 		// 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
@@ -1480,6 +1501,28 @@ func (s3a *S3ApiServer) updateLatestVersionInDirectory(bucket, object, versionId
 	if versionsEntry.Extended == nil {
 		versionsEntry.Extended = make(map[string][]byte)
 	}
+	// Capture the previously-latest version's filename before overwriting
+	// the pointer so the lifecycle engine can stamp NoncurrentSinceNs on
+	// the demoted entry. Same-file overwrites (idempotent retries) are
+	// detected by filename equality and skip the stamp.
+	prevLatestFileName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey])
+
+	// Stamp the demoted entry BEFORE updating the .versions/ directory
+	// pointer. The pointer-flip emits a meta-log event that the
+	// lifecycle router consumes; that router then looks up the demoted
+	// version. If the stamp isn't on the entry yet, the router falls
+	// back to a sibling-mtime derivation that can be arbitrarily wrong
+	// — versioned COPY is the clean break, since the new latest keeps
+	// the source object's mtime instead of recording the demotion
+	// moment. Stamping first closes that race.
+	//
+	// Best-effort: a transient stamp-write failure is logged but not
+	// fatal; the lifecycle engine still falls back to the legacy
+	// derivation in that case.
+	if prevLatestFileName != "" && prevLatestFileName != versionFileName {
+		s3a.markVersionNoncurrent(bucketDir, versionsObjectPath, prevLatestFileName, time.Now().UnixNano())
+	}
+
 	versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(versionId)
 	versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey] = []byte(versionFileName)
 

weed/s3api/s3api_object_versioning.go

@@ -38,6 +38,37 @@ func clearCachedVersionMetadata(extended map[string][]byte) {
 	delete(extended, s3_constants.ExtLatestVersionIsDeleteMarker)
 }
 
+// markVersionNoncurrent stamps ExtNoncurrentSinceNsKey on the named entry
+// inside .versions/. Called when a PUT or delete-marker demotes that entry
+// from current to noncurrent so the s3 lifecycle engine can compute
+// NoncurrentDays due time directly from the stamp instead of deriving it
+// from the next-newer sibling's mtime. demotionNs is captured once per
+// demotion event by the caller (typically time.Now().UnixNano()) and
+// passed in so concurrent demotions on the same object don't race for
+// a wall-clock read inside the helper.
+//
+// Idempotent on retries: if the key is already present, it is overwritten
+// with the new value. Out-of-order overwrites are bounded by the caller's
+// single-timestamp-per-event contract.
+func (s3a *S3ApiServer) markVersionNoncurrent(bucketDir, versionsObjectPath, fileName string, demotionNs int64) {
+	if fileName == "" || demotionNs <= 0 {
+		return
+	}
+	versionsDir := bucketDir + "/" + versionsObjectPath
+	entry, err := s3a.getEntry(versionsDir, fileName)
+	if err != nil {
+		glog.V(2).Infof("markVersionNoncurrent: skip %s/%s: %v", versionsDir, fileName, err)
+		return
+	}
+	if entry.Extended == nil {
+		entry.Extended = make(map[string][]byte)
+	}
+	entry.Extended[s3_constants.ExtNoncurrentSinceNsKey] = []byte(strconv.FormatInt(demotionNs, 10))
+	if err := s3a.updateEntry(versionsDir, entry); err != nil {
+		glog.V(2).Infof("markVersionNoncurrent: update %s/%s: %v", versionsDir, fileName, err)
+	}
+}
+
 // setCachedListMetadata caches list metadata in the .versions directory entry for single-scan efficiency
 func setCachedListMetadata(versionsEntry, versionEntry *filer_pb.Entry) {
 	if versionEntry == nil || versionsEntry == nil {

weed/s3api/s3lifecycle/noncurrent_since.go

@@ -0,0 +1,37 @@
+package s3lifecycle
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+// SuccessorFromEntryStamp returns the explicit noncurrent-since timestamp
+// written by the S3 PUT handler at demotion time
+// (s3_constants.ExtNoncurrentSinceNsKey). Returns zero time if the stamp
+// is missing or unparseable — the caller falls back to derived values
+// (sibling mtime or entry mtime) for legacy entries written before the
+// stamp existed.
+//
+// The lifecycle engine prefers this stamp over the legacy
+// "next-newer sibling mtime" derivation because it records the exact
+// moment a version was demoted, immune to later mtime edits on the
+// sibling. The router and bootstrap walker both read it through this
+// helper so the parsing rules — non-positive, unparseable, missing —
+// stay in one place.
+func SuccessorFromEntryStamp(entry *filer_pb.Entry) time.Time {
+	if entry == nil {
+		return time.Time{}
+	}
+	raw, ok := entry.Extended[s3_constants.ExtNoncurrentSinceNsKey]
+	if !ok || len(raw) == 0 {
+		return time.Time{}
+	}
+	ns, err := strconv.ParseInt(string(raw), 10, 64)
+	if err != nil || ns <= 0 {
+		return time.Time{}
+	}
+	return time.Unix(0, ns)
+}

weed/s3api/s3lifecycle/noncurrent_since_test.go

@@ -0,0 +1,75 @@
+package s3lifecycle
+
+import (
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Direct coverage for the parser. The router and the bootstrap walker
+// both go through this single helper to read the demotion stamp, so
+// pinning every parsing branch here keeps that contract enforced even
+// as the call sites evolve.
+
+func TestSuccessorFromEntryStamp_NilOrMissingReturnsZero(t *testing.T) {
+	assert.True(t, SuccessorFromEntryStamp(nil).IsZero())
+	assert.True(t, SuccessorFromEntryStamp(&filer_pb.Entry{}).IsZero())
+	assert.True(t, SuccessorFromEntryStamp(&filer_pb.Entry{
+		Extended: map[string][]byte{},
+	}).IsZero())
+}
+
+func TestSuccessorFromEntryStamp_EmptyOrInvalidReturnsZero(t *testing.T) {
+	for _, raw := range [][]byte{nil, []byte(""), []byte("nope")} {
+		got := SuccessorFromEntryStamp(&filer_pb.Entry{
+			Extended: map[string][]byte{s3_constants.ExtNoncurrentSinceNsKey: raw},
+		})
+		assert.True(t, got.IsZero(), "value %q must produce zero time", string(raw))
+	}
+}
+
+func TestSuccessorFromEntryStamp_NonPositiveReturnsZero(t *testing.T) {
+	// Stamps are wall-clock UnixNano captured at demotion time; <=0
+	// signals "not set" — caller falls back to derived sibling mtime.
+	for _, raw := range []string{"0", "-1"} {
+		got := SuccessorFromEntryStamp(&filer_pb.Entry{
+			Extended: map[string][]byte{s3_constants.ExtNoncurrentSinceNsKey: []byte(raw)},
+		})
+		assert.True(t, got.IsZero(), "value %q must produce zero time", raw)
+	}
+}
+
+func TestSuccessorFromEntryStamp_PositiveNanosRoundTrip(t *testing.T) {
+	const ns int64 = 1700000000_123456789
+	got := SuccessorFromEntryStamp(&filer_pb.Entry{
+		Extended: map[string][]byte{s3_constants.ExtNoncurrentSinceNsKey: []byte("1700000000123456789")},
+	})
+	assert.Equal(t, time.Unix(0, ns).UTC(), got.UTC())
+}
+
+func TestSuccessorFromEntryStamp_OrderedNanosStayOrdered(t *testing.T) {
+	// Pins the read contract: if two stamps were written in increasing
+	// nanosecond order, the parser must return Times in the same order.
+	// Uses fixed nanosecond values rather than time.Now().UnixNano() —
+	// UnixNano() drops the monotonic component, so back-to-back calls
+	// can decrease if the wall clock steps backward and the property
+	// being verified would then exercise OS clock behavior, not the
+	// parser.
+	require := require.New(t)
+	const earlier int64 = 1700000000_000000001
+	const later int64 = 1700000000_000000002
+
+	earlierEntry := &filer_pb.Entry{
+		Extended: map[string][]byte{s3_constants.ExtNoncurrentSinceNsKey: []byte(strconv.FormatInt(earlier, 10))},
+	}
+	laterEntry := &filer_pb.Entry{
+		Extended: map[string][]byte{s3_constants.ExtNoncurrentSinceNsKey: []byte(strconv.FormatInt(later, 10))},
+	}
+	require.False(SuccessorFromEntryStamp(earlierEntry).After(SuccessorFromEntryStamp(laterEntry)),
+		"parser must preserve write-order between two stamps")
+}

weed/s3api/s3lifecycle/router/router.go

@@ -310,6 +310,14 @@ func routePointerTransitionDisplaced(ctx context.Context, snap *engine.Snapshot,
 	if displaced == nil || displaced.Attributes == nil {
 		return nil
 	}
+	// Prefer the explicit demotion stamp on the displaced entry over the
+	// container-derived successor. Stamp is written by the S3 PUT handler
+	// at the moment the pointer flipped; container value is derived from
+	// the new latest's mtime and may drift across retries.
+	effectiveSuccessor := successor
+	if stamp := s3lifecycle.SuccessorFromEntryStamp(displaced); !stamp.IsZero() {
+		effectiveSuccessor = stamp
+	}
 	idx := 0
 	info := &s3lifecycle.ObjectInfo{
 		Key:              logical,
@@ -318,12 +326,12 @@ func routePointerTransitionDisplaced(ctx context.Context, snap *engine.Snapshot,
 		IsLatest:         false,
 		IsDeleteMarker:   string(displaced.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
 		NoncurrentIndex:  &idx,
-		SuccessorModTime: successor,
+		SuccessorModTime: effectiveSuccessor,
 	}
 	if tags := extractTags(displaced.Extended); len(tags) > 0 {
 		info.Tags = tags
 	}
-	return emitNoncurrentMatches(snap, ev, keys, info, displaced, displacedID, successor)
+	return emitNoncurrentMatches(snap, ev, keys, info, displaced, displacedID, effectiveSuccessor)
 }
 
 // routePointerTransitionExpand routes only the versions that newly
@@ -445,6 +453,12 @@ func routePointerTransitionExpand(ctx context.Context, snap *engine.Snapshot, ev
 		} else {
 			thisSuccessor = successor
 		}
+		// Override with the explicit demotion stamp when present —
+		// PUT-time wall clock beats derived sibling mtime for accuracy
+		// and is immune to mtime edits on the sibling itself.
+		if stamp := s3lifecycle.SuccessorFromEntryStamp(s.entry); !stamp.IsZero() {
+			thisSuccessor = stamp
+		}
 		idx := rank
 		info := &s3lifecycle.ObjectInfo{
 			Key:              logical,

weed/s3api/s3lifecycle/scheduler/bootstrap.go

@@ -327,8 +327,12 @@ func (b *BucketBootstrapper) expandVersionsDir(ctx context.Context, bucket, root
 	}
 	count := 0
 	for i, it := range items {
-		var successor time.Time
-		if i > 0 {
+		// Prefer the explicit demotion stamp written by the S3 PUT
+		// handler. Falling back to the next-newer sibling's mtime is
+		// the legacy derivation and stays in place for entries written
+		// before the stamp was introduced.
+		successor := s3lifecycle.SuccessorFromEntryStamp(it.entry)
+		if successor.IsZero() && i > 0 {
 			prev := items[i-1].entry.Attributes
 			successor = time.Unix(prev.Mtime, int64(prev.MtimeNs))
 		}

weed/s3api/s3lifecycle/scheduler/bootstrap_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"sort"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -648,6 +649,111 @@ func TestExpandVersionsDir_LatestAndNoncurrentsByMtime(t *testing.T) {
 	assert.Equal(t, v2mt.Unix(), byID["v1"].SuccessorModTime.Unix())
 }
 
+func TestExpandVersionsDir_NoncurrentSinceStampOverridesSiblingMtime(t *testing.T) {
+	// When a version entry carries an explicit ExtNoncurrentSinceNsKey
+	// stamp written by the S3 PUT handler at demotion time, that stamp
+	// must take precedence over the legacy "use the next-newer sibling's
+	// mtime" derivation. The stamp records exactly when the version
+	// became noncurrent, which the sibling mtime only approximates.
+	now := time.Now()
+	v1mt := now.Add(-10 * time.Hour) // entry mtime, very old
+	v2mt := now.Add(-1 * time.Hour)  // sibling that would normally drive successor
+	// Stamp v1's demotion at a fixed wall-clock that doesn't match v2.mt.
+	v1demotion := now.Add(-3 * time.Hour)
+
+	v1 := versionFile("v1", v1mt, false)
+	v1.Extended[s3_constants.ExtNoncurrentSinceNsKey] = []byte(
+		strconv.FormatInt(v1demotion.UnixNano(), 10),
+	)
+	v2 := versionFile("v2", v2mt, false)
+
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v2"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {v1, v2},
+		},
+	}
+	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() {
+		require.NotNil(t, ev.BootstrapVersion)
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Contains(t, byID, "v1")
+	// SuccessorModTime must come from the stamp, not from v2's mtime.
+	assert.Equal(t, v1demotion.UnixNano(), byID["v1"].SuccessorModTime.UnixNano(),
+		"stamp must win over sibling mtime; got %v want %v",
+		byID["v1"].SuccessorModTime, v1demotion)
+	assert.NotEqual(t, v2mt.Unix(), byID["v1"].SuccessorModTime.Unix(),
+		"sibling mtime must not be the SuccessorModTime source when stamp is present")
+}
+
+func TestExpandVersionsDir_MissingStampFallsBackToSiblingMtime(t *testing.T) {
+	// Legacy/pre-Phase-1 entries have no stamp. Behavior must be
+	// unchanged: SuccessorModTime falls back to the next-newer sibling's
+	// mtime. This pins the backward-compat path the design promises.
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-1 * time.Hour)
+	v1 := versionFile("v1", v1mt, false) // no stamp
+	v2 := versionFile("v2", v2mt, false)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v2"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {v1, v2},
+		},
+	}
+	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.Equal(t, v2mt.Unix(), byID["v1"].SuccessorModTime.Unix(),
+		"missing stamp must fall through to sibling mtime")
+}
+
+func TestExpandVersionsDir_InvalidStampFallsBackToSiblingMtime(t *testing.T) {
+	// A malformed stamp value is the writer's bug — the reader must
+	// ignore it and fall back to the legacy derivation rather than
+	// blowing up or surfacing a nonsense time.
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-1 * time.Hour)
+	v1 := versionFile("v1", v1mt, false)
+	v1.Extended[s3_constants.ExtNoncurrentSinceNsKey] = []byte("not-a-number")
+	v2 := versionFile("v2", v2mt, false)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v2"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {v1, v2},
+		},
+	}
+	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.Equal(t, v2mt.Unix(), byID["v1"].SuccessorModTime.Unix(),
+		"unparseable stamp must fall through to sibling mtime")
+}
+
 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