s3 lifecycle: walker deletes carry the listed identity and must target the latest version (#12452)

* s3 lifecycle: send the listed entry's identity on walker deletes

The daily walker dispatched LifecycleDelete with no ExpectedIdentity, so
the server skipped its compare-and-swap and an object rewritten between
listing and dispatch was deleted. The list adapter now records each
listed entry's identity, computed the same way as on the event path, and
the walker sends it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3 lifecycle: current-version expiration must target the latest version

The walker addresses the latest version of a versioned key by id. A
version file never changes, so its identity still matches after a newer
PUT, and the delete went ahead: a delete marker over the new version with
versioning enabled, or the new null version deleted while suspended.
LifecycleDelete now resolves the key's latest version for
EXPIRATION_DAYS/EXPIRATION_DATE requests that carry a version id and
answers NOOP_RESOLVED (VERSION_NOT_LATEST) when it is a different one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3 lifecycle: check the latest version for unaddressed suspended expirations

With versioning suspended, an expiration without a version id reads the
plain entry, which is the null version but not necessarily the latest:
a newer real version can sit in .versions. The latest-version check now
runs in the expiration branch, by versioning state: for every suspended
request (an unaddressed one addresses the null version) and for enabled
requests that carry a version id.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3 lifecycle: test that each version sibling carries its own identity

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/s3api/s3api_internal_lifecycle.go

@@ -73,6 +73,12 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 			}
 			return retryLater("TRANSPORT_ERROR: versioning lookup: " + vErr.Error()), nil
 		}
+		// An unaddressed read already resolves the latest when enabled.
+		if state == s3_constants.VersioningSuspended || (state == s3_constants.VersioningEnabled && req.VersionId != "") {
+			if resp := s3a.lifecycleRequireLatest(req, entry); resp != nil {
+				return resp, nil
+			}
+		}
 		switch state {
 		case s3_constants.VersioningEnabled:
 			if _, err := s3a.createDeleteMarker(req.Bucket, req.ObjectPath); err != nil {
@@ -210,6 +216,33 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 	}
 }
 
+// lifecycleRequireLatest refuses a current-version expiration whose version
+// is no longer the key's latest. A version file never changes, so the identity
+// check cannot see a newer write.
+func (s3a *S3ApiServer) lifecycleRequireLatest(req *s3_lifecycle_pb.LifecycleDeleteRequest, entry *filer_pb.Entry) *s3_lifecycle_pb.LifecycleDeleteResponse {
+	latest, err := s3a.getLatestObjectVersion(req.Bucket, req.ObjectPath)
+	if err != nil {
+		if isObjectEntryMiss(err) {
+			return noopResolved("NOT_FOUND")
+		}
+		return retryLater("TRANSPORT_ERROR: latest-version lookup: " + err.Error())
+	}
+	versionIdOrNull := func(e *filer_pb.Entry) string {
+		if id := string(e.Extended[s3_constants.ExtVersionIdKey]); id != "" {
+			return id
+		}
+		return "null"
+	}
+	addressed := req.VersionId
+	if addressed == "" {
+		addressed = versionIdOrNull(entry)
+	}
+	if versionIdOrNull(latest) != addressed {
+		return noopResolved("VERSION_NOT_LATEST")
+	}
+	return nil
+}
+
 func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
 	// req.ObjectPath is `.uploads/<upload_id>` (set by the router from the
 	// init directory's bucket-relative path); reject anything that isn't

weed/s3api/s3api_internal_lifecycle_test.go

@@ -5,16 +5,20 @@ import (
 	"context"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestComputeEntryIdentity_BasicFields(t *testing.T) {
@@ -377,3 +381,90 @@ func TestLifecycleDelete_NoSigningKey_LocalPeerOnly(t *testing.T) {
 		t.Fatalf("expected BLOCKED for empty request without signing key, got %v", resp)
 	}
 }
+
+func TestLifecycleDelete_CurrentExpirationRequiresLatestVersion(t *testing.T) {
+	listed := &filer_pb.Entry{
+		Name:       "v_old",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1_700_000_000, FileSize: 3},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "3,01637037d6", Size: 3}},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte("old")},
+	}
+	newer := &filer_pb.Entry{
+		Name:       "v_new",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1_700_000_100, FileSize: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "3,02637037d6", Size: 5}},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte("new")},
+	}
+	versionsDir := func(latestId string) *filer_pb.Entry {
+		d := &filer_pb.Entry{Name: "obj" + s3_constants.VersionsFolder, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}
+		if latestId != "" {
+			d.Extended = map[string][]byte{
+				s3_constants.ExtLatestVersionIdKey:       []byte(latestId),
+				s3_constants.ExtLatestVersionFileNameKey: []byte("v_" + latestId),
+			}
+		}
+		return d
+	}
+	nullOverwrite := &filer_pb.Entry{
+		Name:       "obj",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1_700_000_200, FileSize: 7},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "3,03637037d6", Size: 7}},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte("null")},
+	}
+	cases := []struct {
+		name       string
+		versioning string
+		versionId  string
+		listed     *filer_pb.Entry
+		entries    map[string]*filer_pb.Entry
+		outcome    s3_lifecycle_pb.LifecycleDeleteOutcome
+	}{
+		{"newer version", s3_constants.VersioningEnabled, "old", listed, map[string]*filer_pb.Entry{
+			"/buckets/b/obj.versions":       versionsDir("new"),
+			"/buckets/b/obj.versions/v_old": listed,
+			"/buckets/b/obj.versions/v_new": newer,
+		}, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED},
+		{"null version written while suspended", s3_constants.VersioningSuspended, "old", listed, map[string]*filer_pb.Entry{
+			"/buckets/b/obj":                nullOverwrite,
+			"/buckets/b/obj.versions":       versionsDir(""),
+			"/buckets/b/obj.versions/v_old": listed,
+		}, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED},
+		{"unaddressed null behind a newer version while suspended", s3_constants.VersioningSuspended, "", nullOverwrite, map[string]*filer_pb.Entry{
+			"/buckets/b/obj":                nullOverwrite,
+			"/buckets/b/obj.versions":       versionsDir("new"),
+			"/buckets/b/obj.versions/v_new": newer,
+		}, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED},
+		{"still latest", s3_constants.VersioningEnabled, "old", listed, map[string]*filer_pb.Entry{
+			"/buckets/b/obj.versions":       versionsDir("old"),
+			"/buckets/b/obj.versions/v_old": listed,
+		}, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE},
+	}
+	for _, kind := range []s3_lifecycle_pb.ActionKind{s3_lifecycle_pb.ActionKind_EXPIRATION_DAYS, s3_lifecycle_pb.ActionKind_EXPIRATION_DATE} {
+		for _, c := range cases {
+			t.Run(kind.String()+"/"+c.name, func(t *testing.T) {
+				entries := map[string]*filer_pb.Entry{}
+				for k, v := range c.entries {
+					entries[k] = proto.Clone(v).(*filer_pb.Entry)
+				}
+				fake := &ambiguousPutFiler{entries: entries, apply: true}
+				s3a := newPutTestServer(t, startFakeFiler(t, fake))
+				s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+				s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: c.versioning})
+				resp, err := s3a.LifecycleDelete(s3LocalPeerCtx(), &s3_lifecycle_pb.LifecycleDeleteRequest{
+					Bucket:           "b",
+					ObjectPath:       "obj",
+					VersionId:        c.versionId,
+					ActionKind:       kind,
+					ExpectedIdentity: computeEntryIdentity(c.listed),
+				})
+				require.NoError(t, err)
+				require.Equal(t, c.outcome, resp.Outcome, resp.Reason)
+				if c.outcome == s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED {
+					require.Equal(t, "VERSION_NOT_LATEST", resp.Reason)
+					require.Zero(t, fake.txnCount, "no delete marker or delete may be written")
+					require.Len(t, fake.entries, len(c.entries))
+				}
+			})
+		}
+	}
+}

weed/s3api/s3lifecycle/bootstrap/walker.go

@@ -15,6 +15,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 )
 
@@ -47,6 +48,10 @@ type Entry struct {
 	// doesn't use it, but the Dispatcher needs it to address the
 	// right version on LifecycleDelete.
 	VersionID string
+
+	// Identity is the listed entry's fingerprint; the server refuses the
+	// delete when the live entry no longer matches it.
+	Identity *router.EntryIdentity
 }
 
 // ListFunc must skip entries with Path <= start so kill-resume picks up

weed/s3api/s3lifecycle/dailyrun/filer_list_func.go

@@ -13,6 +13,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
@@ -143,6 +144,7 @@ func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir
 			Size:     int64(e.Attributes.FileSize),
 			IsLatest: true, // Non-versioned default.
 			Tags:     extractTags(e.Extended),
+			Identity: router.BuildIdentityFromEntry(e),
 		}
 		return cb(entry)
 	})
@@ -273,6 +275,7 @@ func expandVersionsDir(ctx context.Context, client filer_pb.SeaweedFilerClient,
 			NumVersions:      len(items),
 			SuccessorModTime: successor,
 			Tags:             extractTags(it.entry.Extended),
+			Identity:         router.BuildIdentityFromEntry(it.entry),
 		}
 		if !isLatest {
 			rank := i

weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
@@ -570,6 +571,33 @@ func TestFilerListFunc_AttributesPropagate(t *testing.T) {
 	assert.True(t, got.IsLatest)
 }
 
+func TestFilerListFunc_EntriesCarryListedIdentity(t *testing.T) {
+	mtime := time.Date(2026, 5, 11, 12, 0, 0, 1234, time.UTC)
+	flat := fileWithExt("flat", mtime, 4096, map[string][]byte{"k": []byte("v")})
+	flat.Chunks = []*filer_pb.FileChunk{{FileId: "3,01637037d6", Size: 4096}}
+	latest := fileWithExt("v1", mtime, 8, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")})
+	latest.Chunks = []*filer_pb.FileChunk{{FileId: "4,02a1b2c3d4", Size: 8}}
+	older := fileWithExt("v0", mtime.Add(-time.Minute), 9, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v0")})
+	older.Chunks = []*filer_pb.FileChunk{{FileId: "5,03a1b2c3d4", Size: 9}}
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {flat, versionsDir("foo"+s3_constants.VersionsFolder, "v1")},
+		"/buckets/bkt/foo" + s3_constants.VersionsFolder: {latest, older},
+	}}
+	listFn := FilerListFunc(client, "/buckets", nil)
+	byKey := map[string]*bootstrap.Entry{}
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		byKey[e.Path+"@"+e.VersionID] = e
+		return nil
+	}))
+	require.Len(t, byKey, 3)
+	assert.Equal(t, router.BuildIdentityFromEntry(flat), byKey["flat@"].Identity)
+	assert.Equal(t, "3,01637037d6", byKey["flat@"].Identity.HeadFid)
+	assert.Equal(t, router.BuildIdentityFromEntry(latest), byKey["foo@v1"].Identity)
+	assert.Equal(t, "4,02a1b2c3d4", byKey["foo@v1"].Identity.HeadFid)
+	assert.Equal(t, router.BuildIdentityFromEntry(older), byKey["foo@v0"].Identity)
+	assert.Equal(t, "5,03a1b2c3d4", byKey["foo@v0"].Identity.HeadFid)
+}
+
 func TestDirInRuleScope(t *testing.T) {
 	cases := []struct {
 		prefixes []string

weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go

@@ -14,10 +14,9 @@ import (
 
 // WalkerDispatcher adapts LifecycleClient to bootstrap.Dispatcher so
 // the Phase 4b walker can drive the same LifecycleDelete RPC the
-// meta-log replay path uses. No CAS witness is supplied; the server's
-// identityMatches treats nil ExpectedIdentity as "bootstrap call, skip
-// witness" (see weed/s3api/s3api_internal_lifecycle.go), which is the
-// right contract for a full-tree walk that has just observed the entry.
+// meta-log replay path uses, with the listed entry's identity as the
+// CAS witness: an object rewritten between listing and dispatch
+// resolves as STALE_IDENTITY instead of being deleted.
 type WalkerDispatcher struct {
 	Client LifecycleClient
 	// Limiter throttles LifecycleDelete dispatch under the cluster's
@@ -57,13 +56,12 @@ func (d *WalkerDispatcher) Delete(ctx context.Context, action *engine.CompiledAc
 	}
 	rh := action.Key.RuleHash
 	req := &s3_lifecycle_pb.LifecycleDeleteRequest{
-		Bucket:     action.Bucket,
-		ObjectPath: objectPath,
-		VersionId:  entry.VersionID,
-		RuleHash:   rh[:],
-		ActionKind: toProtoActionKind(action.Key.ActionKind),
-		// ExpectedIdentity intentionally nil; server bootstraps from
-		// the live entry on this code path.
+		Bucket:           action.Bucket,
+		ObjectPath:       objectPath,
+		VersionId:        entry.VersionID,
+		RuleHash:         rh[:],
+		ActionKind:       toProtoActionKind(action.Key.ActionKind),
+		ExpectedIdentity: toProtoIdentity(entry.Identity),
 	}
 	kindLabel := action.Key.ActionKind.String()
 	if d.Limiter != nil {

weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/time/rate"
@@ -53,7 +54,8 @@ func TestWalkerDispatcher_NonVersionedSendsExpectedRequest(t *testing.T) {
 	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
 	d := &WalkerDispatcher{Client: c}
 	a := sampleAction(t, s3lifecycle.ActionKindExpirationDays)
-	err := d.Delete(context.Background(), a, &bootstrap.Entry{Path: "obj"})
+	identity := &router.EntryIdentity{MtimeNs: 1_700_000_000_123, Size: 42, HeadFid: "3,01637037d6", ExtendedHash: []byte{0xab}}
+	err := d.Delete(context.Background(), a, &bootstrap.Entry{Path: "obj", Identity: identity})
 	require.NoError(t, err)
 	require.NotNil(t, c.lastReq)
 	assert.Equal(t, "bkt", c.lastReq.Bucket)
@@ -61,8 +63,11 @@ func TestWalkerDispatcher_NonVersionedSendsExpectedRequest(t *testing.T) {
 	assert.Equal(t, "", c.lastReq.VersionId)
 	assert.Equal(t, a.Key.RuleHash[:], c.lastReq.RuleHash)
 	assert.Equal(t, s3_lifecycle_pb.ActionKind_EXPIRATION_DAYS, c.lastReq.ActionKind)
-	// Bootstrap-style call: server skips CAS witness when nil.
-	assert.Nil(t, c.lastReq.ExpectedIdentity)
+	require.NotNil(t, c.lastReq.ExpectedIdentity, "the listed entry's identity is the CAS witness")
+	assert.Equal(t, identity.MtimeNs, c.lastReq.ExpectedIdentity.MtimeNs)
+	assert.Equal(t, identity.Size, c.lastReq.ExpectedIdentity.Size)
+	assert.Equal(t, identity.HeadFid, c.lastReq.ExpectedIdentity.HeadFid)
+	assert.Equal(t, identity.ExtendedHash, c.lastReq.ExpectedIdentity.ExtendedHash)
 }
 
 func TestWalkerDispatcher_VersionedPassesVersionID(t *testing.T) {

weed/s3api/s3lifecycle/router/router.go

@@ -491,7 +491,7 @@ func routePointerTransitionExpand(ctx context.Context, snap *engine.Snapshot, ev
 // single-lookup and full-expansion paths.
 func emitNoncurrentMatches(snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, info *s3lifecycle.ObjectInfo, entry *filer_pb.Entry, versionID string, successor time.Time) []Match {
 	eventTime := time.Unix(0, ev.TsNs)
-	identity := buildIdentityFromEntry(entry)
+	identity := BuildIdentityFromEntry(entry)
 	var matches []Match
 	for _, key := range keys {
 		if key.ActionKind != s3lifecycle.ActionKindNoncurrentDays && key.ActionKind != s3lifecycle.ActionKindNewerNoncurrent {
@@ -561,7 +561,7 @@ func routeVersionFileExpiration(snap *engine.Snapshot, ev *reader.Event, keys []
 		info.Tags = tags
 	}
 	eventTime := time.Unix(0, ev.TsNs)
-	identity := buildIdentityFromEntry(entry)
+	identity := BuildIdentityFromEntry(entry)
 	var matches []Match
 	for _, key := range keys {
 		action := snap.Action(key)
@@ -643,7 +643,7 @@ func routeSoleSurvivorMarker(ctx context.Context, snap *engine.Snapshot, ev *rea
 		info.Tags = tags
 	}
 	eventTime := time.Unix(0, ev.TsNs)
-	identity := buildIdentityFromEntry(entry)
+	identity := BuildIdentityFromEntry(entry)
 	var matches []Match
 	for _, key := range keys {
 		if key.ActionKind != s3lifecycle.ActionKindExpiredDeleteMarker {
@@ -703,7 +703,7 @@ func routeBootstrapVersion(snap *engine.Snapshot, ev *reader.Event, keys []s3lif
 		info.Tags = tags
 	}
 	eventTime := time.Unix(0, ev.TsNs)
-	identity := buildIdentityFromEntry(entry)
+	identity := BuildIdentityFromEntry(entry)
 	var matches []Match
 	for _, key := range keys {
 		action := snap.Action(key)
@@ -873,10 +873,10 @@ func mpuInitInfo(ev *reader.Event, entry *filer_pb.Entry) (destKey string, ok bo
 // buildIdentity captures the entry's schedule-time fingerprint for the CAS
 // witness. Returns nil if the event has no entry to fingerprint (deletes).
 func buildIdentity(ev *reader.Event) *EntryIdentity {
-	return buildIdentityFromEntry(ev.NewEntry)
+	return BuildIdentityFromEntry(ev.NewEntry)
 }
 
-func buildIdentityFromEntry(entry *filer_pb.Entry) *EntryIdentity {
+func BuildIdentityFromEntry(entry *filer_pb.Entry) *EntryIdentity {
 	if entry == nil {
 		return nil
 	}