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 }