From 2be4498c045a9735cad17339ac92c41f7b9bb9a7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 09:39:12 -0800 Subject: [PATCH] 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 * 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 * 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 * s3 lifecycle: test that each version sibling carries its own identity Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3api_internal_lifecycle.go | 33 +++++++ weed/s3api/s3api_internal_lifecycle_test.go | 91 +++++++++++++++++++ weed/s3api/s3lifecycle/bootstrap/walker.go | 5 + .../s3lifecycle/dailyrun/filer_list_func.go | 3 + .../dailyrun/filer_list_func_test.go | 28 ++++++ .../s3lifecycle/dailyrun/walker_dispatcher.go | 20 ++-- .../dailyrun/walker_dispatcher_test.go | 11 ++- weed/s3api/s3lifecycle/router/router.go | 12 +-- 8 files changed, 183 insertions(+), 20 deletions(-) diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index 81b42d0501..b60da67871 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/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/` (set by the router from the // init directory's bucket-relative path); reject anything that isn't diff --git a/weed/s3api/s3api_internal_lifecycle_test.go b/weed/s3api/s3api_internal_lifecycle_test.go index 54d44a49e5..4b97958fa8 100644 --- a/weed/s3api/s3api_internal_lifecycle_test.go +++ b/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)) + } + }) + } + } +} diff --git a/weed/s3api/s3lifecycle/bootstrap/walker.go b/weed/s3api/s3lifecycle/bootstrap/walker.go index ac36798728..845a459d5a 100644 --- a/weed/s3api/s3lifecycle/bootstrap/walker.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go index 0a24d5c518..f646564ecf 100644 --- a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go index aa94024356..b5f5964f74 100644 --- a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go index eaff959b8c..2ab26162ba 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go +++ b/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 { diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go index f30239394a..e862179512 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go +++ b/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) { diff --git a/weed/s3api/s3lifecycle/router/router.go b/weed/s3api/s3lifecycle/router/router.go index 81c084e5c1..fc7769becd 100644 --- a/weed/s3api/s3lifecycle/router/router.go +++ b/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 }