From c6ad6dcf74ac4383b5d48bf173dac64c91de7981 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 8 May 2026 23:24:08 -0700 Subject: [PATCH] feat(s3/lifecycle): sole-survivor delete-marker routing (Phase 5b/2) (#9381) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): sole-survivor delete-marker routing (Phase 5b/2) Production stores delete markers under .versions/; buildObjectInfo skips every version-folder event because it can't tell current from noncurrent without sibling state. EXP_DM is the one rule that can be routed from the file path alone: if the marker is the only entry under .versions// then it's necessarily the latest. Add SiblingLister; gate the listing on (versioned, version-folder path, delete-marker entry, EXP_DM rule active for the bucket, lister supplied) so non-marker version writes pay nothing. The Match carries the LOGICAL key in ObjectKey and the marker's version_id in VersionID so the dispatcher can reach deleteSpecificObjectVersion(bucket, logical, vid); without a version_id the dispatcher would BLOCK and freeze the cursor. Server-side dispatch re-checks sole-survivor before deleting: another PUT can land between schedule and dispatch, and identity-CAS only covers the marker entry, not the directory shape. Lists .versions// with limit 2 and returns NOOP_RESOLVED if count != 1 or the surviving entry is not the same version. Fix isDeleteMarkerEntry to compare string(v) == "true" — production writes []byte("true"), not {1}; every other reader of ExtDeleteMarkerKey in this repo uses the string predicate. filerSiblingLister.Count caps at limit 2 — callers only distinguish "sole survivor" (1) from "more than one" (>=2). Follows #9373. * fix(s3/lifecycle): gate sibling listing on active event-driven EXP_DM hasActionKind tested key presence only; an EXP_DM rule whose action was inactive (BootstrapComplete=false) or scan-only would still trigger the sibling-listing RPC even though no match could fire. Replace with a helper that mirrors the per-key filter Route already applies before emitting matches: kind matches, snap.Action returns non-nil, IsActive, and Mode == ModeEventDriven. Add a regression test that builds a versioned snapshot with the rule but no PriorStates (action stays inactive) and asserts the lister is never consulted. * chore(s3/lifecycle): trim verbose comments * fix(s3/lifecycle): null version, hard-delete trigger, latest pointer Three correctness gaps in EXP_DM routing. 1. Null version. SiblingLister.Count saw only .versions//, but a pre-versioning bare-key object survives outside that folder when versioning is enabled later. Marker + null would look like count==1 and EXP_DM would re-expose the old object. Replace Count with Survivors which also reports HasNullVersion; route- and dispatch-time suppress when set. 2. Hard-delete trigger. The router skipped events with NewEntry==nil so a noncurrent hard-delete that left the marker as sole survivor never fired EXP_DM. Allow version-folder hard-deletes through; the lister's LoneEntry carries the surviving marker's version_id and identity. 3. Latest pointer. checkSoleSurvivorMarker only checked count and filename. The worker can race createDeleteMarker between the file write and the .versions/ directory metadata update. Require ExtLatestVersionIdKey == versionId; missing pointer returns retry-later instead of deleting. Adds a null-version exists check on the dispatch path too. * fix(s3/lifecycle): normalize null-version lookup errors and detect dir markers Two correctness bugs in the null-version check. Route-side: filerSiblingLister called client.LookupDirectoryEntry directly. SeaweedList wraps not-found via filer_pb.LookupEntry, which normalizes the gRPC string-mapped not-found into ErrNotFound. The raw client returns it as a generic error instead, so every absent null-version (the common case) bubbled up as an error and the router suppressed every otherwise-valid match. Use filer_pb.LookupEntry. Both sides: explicit S3 directory-key markers (object names ending in /) are stored as directory entries with Attributes.Mime set; processExplicitDirectory in the listing path treats them as null versions. The previous check was !IsDirectory only, which let the marker delete proceed and re-expose the directory key. Add IsDirectoryKeyObject() to both predicates. Also use util.NewFullPath(...).DirAndName() for the parent/name split so a trailing-slash key resolves to the same underlying entry path as the listing code. * fix(s3/lifecycle): EXP_DM ctx propagation, nil-entry guard, fast-path skip Three small follow-ups on the EXP_DM dispatch path. checkSoleSurvivorMarker now takes ctx instead of context.Background() so worker shutdown / deadlines cancel the SeaweedList RPC instead of stalling. If SeaweedList fires the lone callback with entry==nil, firstName stays empty and the marker-replaced check would short-circuit; that's the one shape that bypasses the dispatch guard, so retry-later instead. routeSoleSurvivorMarker now skips the Survivors RPC on regular non-marker version creates — those always have Count >= 2, so the listing was wasted load on every versioned write under an EXP_DM rule. Hard-delete events (NewEntry==nil) and marker creates still flow through. Added a regression test asserting the regular-create case doesn't consult the lister. Documented that logicalKeyFromVersionPath rejects bucket-root markers intentionally. --- weed/s3api/s3api_internal_lifecycle.go | 95 +++++ weed/s3api/s3lifecycle/dispatcher/pipeline.go | 4 +- .../s3lifecycle/dispatcher/pipeline_test.go | 2 +- .../s3lifecycle/dispatcher/sibling_lister.go | 61 ++++ weed/s3api/s3lifecycle/router/router.go | 225 +++++++++--- weed/s3api/s3lifecycle/router/router_test.go | 326 ++++++++++++++++-- 6 files changed, 629 insertions(+), 84 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dispatcher/sibling_lister.go diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index dff87bafe..42c76c842 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -120,6 +120,15 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle return noopResolved("VERSION_IS_LATEST"), nil } } + // Re-check sole-survivor: a fresh PUT can land between schedule + // and dispatch. Identity-CAS upstream covers the marker bytes; + // this covers the directory shape. + if req.ActionKind == s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER { + outcome, err := s3a.checkSoleSurvivorMarker(ctx, req.Bucket, req.ObjectPath, req.VersionId) + if outcome != nil || err != nil { + return outcome, err + } + } if err := s3a.deleteSpecificObjectVersion(req.Bucket, req.ObjectPath, req.VersionId); err != nil { if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrObjectNotFound) { return noopResolved("NOT_FOUND_AT_DELETE"), nil @@ -175,6 +184,92 @@ func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle return done(), nil } +// checkSoleSurvivorMarker returns nil to proceed with the delete, or a +// terminal response when state has drifted: count != 1, the surviving +// entry is a different version, the .versions/ directory's latest +// pointer doesn't name versionId, or a bare null-version exists outside +// .versions/. Pointer missing while a marker is present is treated as +// retry-later — the create races with the directory metadata update. +func (s3a *S3ApiServer) checkSoleSurvivorMarker(ctx context.Context, bucket, object, versionId string) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + bucketDir := s3a.bucketDir(bucket) + versionsDir := bucketDir + "/" + object + s3_constants.VersionsFolder + count := 0 + var firstName string + err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return filer_pb.SeaweedList(ctx, client, versionsDir, "", func(entry *filer_pb.Entry, _ bool) error { + count++ + if count == 1 && entry != nil { + firstName = entry.Name + } + return nil + }, "", false, 2) + }) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return noopResolved("NOT_FOUND"), nil + } + return retryLater("TRANSPORT_ERROR: sole-survivor list: " + err.Error()), nil + } + if count == 0 { + return noopResolved("NOT_FOUND"), nil + } + if count > 1 { + return noopResolved("NOT_SOLE_SURVIVOR"), nil + } + // SeaweedList delivered a single callback but with a nil entry; we + // can't compare names so retry rather than silently bypass the + // marker-replaced check. + if firstName == "" { + return retryLater("PENDING_SURVIVOR_ENTRY"), nil + } + if versionId != "" && firstName != s3a.getVersionFileName(versionId) { + return noopResolved("MARKER_REPLACED"), nil + } + // Latest-pointer check: createDeleteMarker writes the marker file + // and then updates the parent directory's Extended map. Reading + // before the second step lands would see count==1 but no pointer; + // retry-later rather than mistakenly delete. + parent, name := path.Split(versionsDir) + parent = strings.TrimRight(parent, "/") + if parent == "" { + parent = "/" + } + versionsEntry, err := s3a.getEntry(parent, name) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return noopResolved("NOT_FOUND"), nil + } + return retryLater("TRANSPORT_ERROR: latest-pointer lookup: " + err.Error()), nil + } + if versionsEntry == nil { + return noopResolved("NOT_FOUND"), nil + } + latest, hasPointer := versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey] + if !hasPointer || len(latest) == 0 { + return retryLater("PENDING_LATEST_POINTER"), nil + } + if string(latest) != versionId { + return noopResolved("MARKER_NOT_LATEST"), nil + } + // Null-version check: pre-versioning objects survive as the bare + // /. Both regular files and explicit S3 directory-key + // markers (object names ending in /) qualify; the listing path + // (s3api_object_versioning.go processExplicitDirectory) treats both + // as the null version. getEntry uses NewFullPath so a trailing slash + // in object splits the same as a regular key. + bareEntry, err := s3a.getEntry(bucketDir, object) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return nil, nil + } + return retryLater("TRANSPORT_ERROR: null-version lookup: " + err.Error()), nil + } + if bareEntry != nil && (!bareEntry.IsDirectory || bareEntry.IsDirectoryKeyObject()) { + return noopResolved("NULL_VERSION_PRESENT"), nil + } + return nil, nil +} + // isCurrentLatestVersion reports whether versionId is the version the // .versions/ directory currently points to. SeaweedFS records the latest // version on the parent directory's Extended map; without consulting it, diff --git a/weed/s3api/s3lifecycle/dispatcher/pipeline.go b/weed/s3api/s3lifecycle/dispatcher/pipeline.go index d1cf2796f..3c689c72d 100644 --- a/weed/s3api/s3lifecycle/dispatcher/pipeline.go +++ b/weed/s3api/s3lifecycle/dispatcher/pipeline.go @@ -205,6 +205,8 @@ func (p *Pipeline) Run(ctx context.Context) error { cancel() // wake the dispatcher goroutine to drain & exit }() + lister := &filerSiblingLister{client: p.FilerClient, bucketsPath: p.BucketsPath} + // Router/dispatcher goroutine: pulls events, routes them to per-shard // schedules, ticks every shard's dispatcher on the same cadence, and // checkpoints every shard's cursor on the checkpoint cadence. One @@ -257,7 +259,7 @@ func (p *Pipeline) Run(ctx context.Context) error { // against the prior (empty) snapshot would silently drop // every match. Engine.Snapshot is an atomic Load. snap := p.Engine.Snapshot() - for _, m := range router.Route(snap, ev, time.Now()) { + for _, m := range router.Route(runCtx, snap, ev, time.Now(), lister) { st.dispatch.Schedule.Add(m) } case <-dt.C: diff --git a/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go b/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go index aeb5ed1b8..36c7f3628 100644 --- a/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go +++ b/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go @@ -48,7 +48,7 @@ func TestPipelineIntegrationInMemory(t *testing.T) { }, } - matches := router.Route(snap, ev, now) + matches := router.Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match, got %v", matches) } diff --git a/weed/s3api/s3lifecycle/dispatcher/sibling_lister.go b/weed/s3api/s3lifecycle/dispatcher/sibling_lister.go new file mode 100644 index 000000000..bcbfaa000 --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/sibling_lister.go @@ -0,0 +1,61 @@ +package dispatcher + +import ( + "context" + "errors" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// filerSiblingLister inspects the .versions// folder and the bare +// null-version. Listing caps at 2 — callers only distinguish 1 vs >=2. +type filerSiblingLister struct { + client filer_pb.SeaweedFilerClient + bucketsPath string +} + +func (l *filerSiblingLister) Survivors(ctx context.Context, bucket, objectKey string) (router.Survivors, error) { + bucketPath := strings.TrimSuffix(l.bucketsPath, "/") + "/" + bucket + versionsDir := bucketPath + "/" + objectKey + s3_constants.VersionsFolder + + var s router.Survivors + err := filer_pb.SeaweedList(ctx, l.client, versionsDir, "", func(entry *filer_pb.Entry, _ bool) error { + s.Count++ + if s.Count == 1 && entry != nil { + s.LoneEntry = entry + } else if s.Count > 1 { + s.LoneEntry = nil + } + return nil + }, "", false, 2) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + return router.Survivors{}, err + } + + // NewFullPath strips a trailing slash from objectKey so directory-key + // objects (foo/) split the same as regular keys. LookupEntry + // normalizes the gRPC string-mapped not-found into ErrNotFound; a + // raw client.LookupDirectoryEntry would return that as a generic + // error and suppress every otherwise-valid match. + parent, name := util.NewFullPath(bucketPath, objectKey).DirAndName() + resp, err := filer_pb.LookupEntry(ctx, l.client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: parent, + Name: name, + }) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return s, nil + } + return router.Survivors{}, err + } + // Bare regular file or an explicit S3 directory-marker (an empty + // directory entry with Mime set) both count as the null version. + if resp.Entry != nil && (!resp.Entry.IsDirectory || resp.Entry.IsDirectoryKeyObject()) { + s.HasNullVersion = true + } + return s, nil +} diff --git a/weed/s3api/s3lifecycle/router/router.go b/weed/s3api/s3lifecycle/router/router.go index 0c61da313..af1d82ce5 100644 --- a/weed/s3api/s3lifecycle/router/router.go +++ b/weed/s3api/s3lifecycle/router/router.go @@ -1,9 +1,11 @@ package router import ( + "context" "strings" "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" @@ -11,6 +13,22 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" ) +// SiblingLister inspects the surviving versions of a versioned key. +// nil receiver or an error means "unknown" — callers suppress. +type SiblingLister interface { + Survivors(ctx context.Context, bucket, objectKey string) (Survivors, error) +} + +// Survivors describes the state under .versions// plus the bare +// null-version that exists when versioning was turned on after the +// object was first PUT (s3api_object_versioning.go treats / +// as a regular file, the null version, in that case). +type Survivors struct { + Count int // entries under .versions//, capped at 2 + LoneEntry *filer_pb.Entry // populated when Count == 1 + HasNullVersion bool // bare / exists as a regular file +} + // Match is one (event, action) pair where EvaluateAction fired. The // dispatcher runs `LifecycleDelete` at DueTime; identity-CAS in the RPC // guards against drift between schedule time and dispatch time. @@ -42,26 +60,33 @@ type EntryIdentity struct { ExtendedHash []byte } -// Route returns the matches that fire for ev against snap. Only EVENT_DRIVEN -// actions on this bucket are considered; actions in SCAN_AT_DATE or DISABLED -// modes are out-of-band of the event stream. Inactive actions -// (BootstrapComplete=false) are also skipped. -func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match { +// Route returns the matches that fire for ev against snap. Only active +// event-driven actions are considered; SCAN_AT_DATE and DISABLED bypass +// this path. +func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now time.Time, lister SiblingLister) []Match { if snap == nil || ev == nil { return nil } - // Hard deletes carry no schedule-relevant state: an Expiration would - // hit NOOP_RESOLVED at dispatch time anyway, ExpiredObjectDeleteMarker - // only fires on the latest-version delete-marker which is a Create - // from the server's perspective. Skip rather than burn a schedule slot. - if ev.NewEntry == nil { - return nil - } keys := snap.BucketActionKeys(ev.Bucket) if len(keys) == 0 { return nil } - info := buildObjectInfo(ev, snap.BucketVersioned(ev.Bucket)) + versioned := snap.BucketVersioned(ev.Bucket) + + // EXP_DM can fire on two version-folder events: the marker create + // (sole survivor immediately) and a noncurrent hard-delete that + // leaves only the marker behind. Both reach routeSoleSurvivorMarker. + if versioned && isVersionFolderPath(ev.Key) { + if !hasActiveEventDrivenAction(snap, keys, s3lifecycle.ActionKindExpiredDeleteMarker) { + return nil + } + return routeSoleSurvivorMarker(ctx, snap, ev, keys, lister) + } + + if ev.NewEntry == nil { + return nil + } + info := buildObjectInfo(ev, versioned) if info == nil { return nil } @@ -112,42 +137,127 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match { return matches } -// buildObjectInfo derives an ObjectInfo from a meta-log event and the -// bucket's versioning state. Returns nil when the event has no usable -// shape (missing attributes, hard delete already handled upstream). -// -// On a versioned bucket the storage layout (.versions/v_) is -// shared between the current latest and the noncurrent versions; the -// latest pointer lives in the .versions/ directory's Extended map and -// is updated separately. Without that pointer-transition signal here, -// the router conservatively classifies every event as if it were the -// current version (IsLatest=true) so it never deletes the live -// latest; bootstrap walking + the server-side dispatch guard handle -// noncurrent retention. NumVersions=0 keeps ExpiredObjectDeleteMarker -// (which requires sole-survivor) suppressed. -// -// MPU init directories at .uploads/ populate IsMPUInit and -// use the destination object key from the entry's Extended map for -// filter matching, so a rule with Filter.Prefix=foo/ matches an MPU -// uploading to foo/bar.txt. +// routeSoleSurvivorMarker emits an EXP_DM Match against the LOGICAL key +// (so the dispatcher can call deleteSpecificObjectVersion) with the +// marker's version_id. Handles two events: a marker create (the new +// entry IS the marker) and a noncurrent hard-delete that leaves the +// marker behind (the listing's lone entry IS the marker). The server +// re-checks before deleting. +func routeSoleSurvivorMarker(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, lister SiblingLister) []Match { + if lister == nil { + return nil + } + // Skip the listing RPC for events that can't possibly produce a + // sole-survivor marker: a regular non-marker version create/update + // always lands at Count >= 2. + if ev.NewEntry != nil && !isDeleteMarkerEntry(ev.NewEntry) { + return nil + } + logicalKey, ok := logicalKeyFromVersionPath(ev.Key) + if !ok { + return nil + } + s, err := lister.Survivors(ctx, ev.Bucket, logicalKey) + if err != nil { + glog.V(2).Infof("lifecycle router: survivors %s/%s: %v", ev.Bucket, logicalKey, err) + return nil + } + // Pre-versioning bare-key objects (the "null" version) live outside + // .versions/. Treating count==1 as sole-survivor while a null + // version exists would let lifecycle delete the marker and re-expose + // the old object. + if s.Count != 1 || s.HasNullVersion || s.LoneEntry == nil { + return nil + } + if !isDeleteMarkerEntry(s.LoneEntry) { + return nil + } + versionID := string(s.LoneEntry.Extended[s3_constants.ExtVersionIdKey]) + if versionID == "" { + // Empty version_id would BLOCK at dispatch and freeze the cursor. + return nil + } + entry := s.LoneEntry + info := &s3lifecycle.ObjectInfo{ + Key: logicalKey, + ModTime: time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)), + Size: int64(entry.Attributes.FileSize), + IsLatest: true, + IsDeleteMarker: true, + NumVersions: 1, + } + 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 { + if key.ActionKind != s3lifecycle.ActionKindExpiredDeleteMarker { + continue + } + action := snap.Action(key) + if action == nil || !action.IsActive() || action.Mode != engine.ModeEventDriven { + continue + } + dueTime := info.ModTime.Add(action.Delay) + res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime) + if res.Action == s3lifecycle.ActionNone { + continue + } + matches = append(matches, Match{ + Key: key, + Action: action, + Result: res, + EventTs: eventTime, + DueTime: dueTime, + Bucket: ev.Bucket, + ObjectKey: logicalKey, + VersionID: versionID, + Identity: identity, + }) + } + return matches +} + +// logicalKeyFromVersionPath extracts from .versions/. +// Returns false for a bucket-root marker (path = ".versions/"), since +// AWS S3 has no concept of an object at the bucket key itself. +func logicalKeyFromVersionPath(versionPath string) (string, bool) { + lastSlash := strings.LastIndex(versionPath, "/") + if lastSlash <= 0 { + return "", false + } + parent := versionPath[:lastSlash] + if !strings.HasSuffix(parent, s3_constants.VersionsFolder) { + return "", false + } + logical := strings.TrimSuffix(parent, s3_constants.VersionsFolder) + if logical == "" { + return "", false + } + return logical, true +} + +// buildObjectInfo derives an ObjectInfo from a meta-log event. Returns +// nil for shapes the router can't classify safely: missing attributes, +// non-MPU directories, version-folder files (those route through +// routeSoleSurvivorMarker upstream when EXP_DM applies). On a versioned +// bucket the latest pointer lives in the .versions/ directory's +// Extended map; without it we leave NumVersions=0 so the bootstrap walk +// drives noncurrent retention. func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo { entry := ev.NewEntry if entry == nil || entry.Attributes == nil { return nil } if destKey, ok := mpuInitInfo(ev, entry); ok { - // MPU intermediate state has no tags, no versions, no delete-marker - // semantics. info.Key is the user's destination key so a rule's - // Filter.Prefix matches the eventual object; the dispatcher already - // carries .uploads/ in m.ObjectKey for ABORT_MPU. return &s3lifecycle.ObjectInfo{ Key: destKey, ModTime: time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)), IsMPUInit: true, } } - // Directory entries that aren't MPU inits aren't lifecycle subjects: - // the .versions/ folder itself, prefix dirs, etc. — emit nothing. if entry.IsDirectory { return nil } @@ -159,22 +269,9 @@ func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo { NumVersions: 1, } if versioned { - // On a versioned bucket the actual file path doesn't tell us - // whether the entry is the current latest or a noncurrent - // version — the latest pointer lives in the .versions/ - // directory's Extended map and isn't part of this event. We - // also can't compute NumVersions / NoncurrentIndex here. Skip - // any version-folder file event for now; bootstrap walking - // drives noncurrent retention and current-version expiration - // for versioned buckets until pointer-transition routing - // lands. The bare-key path (null-version, pre-versioning - // objects) keeps the regular routing. if isVersionFolderPath(ev.Key) { return nil } - // NumVersions=0 keeps ExpiredObjectDeleteMarker (sole-survivor - // gate) suppressed for the bare-key delete-marker case until - // sibling listing lands. info.NumVersions = 0 } if tags := extractTags(entry.Extended); len(tags) > 0 { @@ -226,7 +323,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 { - entry := ev.NewEntry + return buildIdentityFromEntry(ev.NewEntry) +} + +func buildIdentityFromEntry(entry *filer_pb.Entry) *EntryIdentity { if entry == nil { return nil } @@ -266,10 +366,31 @@ func extractTags(ext map[string][]byte) map[string]string { return out } +// hasActiveEventDrivenAction gates I/O (e.g. sibling listing) on whether +// a match could actually fire. Mirrors the per-key filter in Route so +// disabled or scan-only actions don't pay the RPC. +func hasActiveEventDrivenAction(snap *engine.Snapshot, keys []s3lifecycle.ActionKey, kind s3lifecycle.ActionKind) bool { + for _, k := range keys { + if k.ActionKind != kind { + continue + } + a := snap.Action(k) + if a == nil { + continue + } + if a.IsActive() && a.Mode == engine.ModeEventDriven { + return true + } + } + return false +} + +// isDeleteMarkerEntry mirrors every read site for ExtDeleteMarkerKey: +// production writes []byte("true"). func isDeleteMarkerEntry(entry *filer_pb.Entry) bool { if entry == nil || len(entry.Extended) == 0 { return false } v, ok := entry.Extended[s3_constants.ExtDeleteMarkerKey] - return ok && len(v) == 1 && v[0] == 1 + return ok && string(v) == "true" } diff --git a/weed/s3api/s3lifecycle/router/router_test.go b/weed/s3api/s3lifecycle/router/router_test.go index eb9a96a32..909576f43 100644 --- a/weed/s3api/s3lifecycle/router/router_test.go +++ b/weed/s3api/s3lifecycle/router/router_test.go @@ -1,6 +1,8 @@ package router import ( + "context" + "errors" "testing" "time" @@ -45,7 +47,7 @@ func eventCreate(bucket, key string, modTimeS, size int64, ts int64) *reader.Eve } func TestRouteNoSnapshotNoMatches(t *testing.T) { - if got := Route(nil, eventCreate("bk", "k", 0, 1, 1), time.Now()); got != nil { + if got := Route(context.Background(), nil, eventCreate("bk", "k", 0, 1, 1), time.Now(), nil); got != nil { t.Fatalf("nil snap should yield nil, got %v", got) } } @@ -54,7 +56,7 @@ func TestRouteMissingBucketNoMatches(t *testing.T) { rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1} snap := compileWith(rule, activatedPrior(rule)) ev := eventCreate("other-bucket", "k", 0, 1, 1) - if got := Route(snap, ev, time.Now()); got != nil { + if got := Route(context.Background(), snap, ev, time.Now(), nil); got != nil { t.Fatalf("foreign bucket should yield nil, got %v", got) } } @@ -68,7 +70,7 @@ func TestRouteInactiveSkipped(t *testing.T) { old := now.Add(-48 * time.Hour) // past 1-day expiration ev := eventCreate("bk", "k", old.Unix(), 1, old.UnixNano()) - if got := Route(snap, ev, now); got != nil { + if got := Route(context.Background(), snap, ev, now, nil); got != nil { t.Fatalf("inactive action should not match, got %v", got) } } @@ -81,7 +83,7 @@ func TestRouteExpirationDaysFires(t *testing.T) { old := now.Add(-48 * time.Hour) // past 1-day expiration ev := eventCreate("bk", "k", old.Unix(), 1, old.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match, got %+v", matches) } @@ -107,7 +109,7 @@ func TestRouteFreshObjectSchedulesInFuture(t *testing.T) { now := time.Now() ev := eventCreate("bk", "k", now.Unix(), 1, now.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match (scheduled), got %v", matches) } @@ -127,13 +129,13 @@ func TestRouteRespectsPrefixFilter(t *testing.T) { // Out of prefix: no match. ev := eventCreate("bk", "data/file", old.Unix(), 1, old.UnixNano()) - if got := Route(snap, ev, now); got != nil { + if got := Route(context.Background(), snap, ev, now, nil); got != nil { t.Fatalf("out-of-prefix should not match, got %v", got) } // In prefix: matches. ev = eventCreate("bk", "logs/file", old.Unix(), 1, old.UnixNano()) - if got := Route(snap, ev, now); len(got) != 1 { + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 1 { t.Fatalf("in-prefix should match, got %v", got) } } @@ -153,7 +155,7 @@ func TestRouteSkipsHardDelete(t *testing.T) { Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 1}, }, } - if got := Route(snap, ev, now); got != nil { + if got := Route(context.Background(), snap, ev, now, nil); got != nil { t.Fatalf("hard delete should not route, got %v", got) } } @@ -169,7 +171,7 @@ func TestRouteSkipsMissingAttributes(t *testing.T) { Key: "k", NewEntry: &filer_pb.Entry{Name: "k"}, // no Attributes } - if got := Route(snap, ev, time.Now()); got != nil { + if got := Route(context.Background(), snap, ev, time.Now(), nil); got != nil { t.Fatalf("missing-Attributes event should not route, got %v", got) } } @@ -183,7 +185,7 @@ func TestRouteIdentityCapturedForNewEntry(t *testing.T) { ev := eventCreate("bk", "k", old.Unix(), 42, old.UnixNano()) ev.NewEntry.Chunks = []*filer_pb.FileChunk{{FileId: "1,abc"}} - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match, got %v", matches) } @@ -214,7 +216,7 @@ func TestRouteIdentityHashesExtended(t *testing.T) { "Content-Type": []byte("text/plain"), } - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match, got %v", matches) } @@ -259,7 +261,7 @@ func TestRouteMPUInitFiresAbortAfterDelay(t *testing.T) { init := now.AddDate(0, 0, -8) ev := mpuInitEvent("bk", "u1", "logs/foo.txt", init.Unix(), init.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match, got %v", matches) } @@ -285,7 +287,7 @@ func TestRouteMPUInitFilteredOutByPrefix(t *testing.T) { init := now.AddDate(0, 0, -8) ev := mpuInitEvent("bk", "u2", "data/foo.txt", init.Unix(), init.UnixNano()) - if got := Route(snap, ev, now); len(got) != 0 { + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 { t.Fatalf("expected 0 matches for foreign prefix, got %v", got) } } @@ -313,7 +315,7 @@ func TestRouteMPUInitMissingDestKeySkipped(t *testing.T) { }, } - if got := Route(snap, ev, now); len(got) != 0 { + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 { t.Fatalf("expected 0 matches for missing destKey, got %v", got) } } @@ -341,7 +343,7 @@ func TestRouteMPUPartEventSkipped(t *testing.T) { }, } - if got := Route(snap, ev, now); len(got) != 0 { + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 { t.Fatalf("expected 0 matches for part event, got %v", got) } } @@ -362,7 +364,7 @@ func TestRouteMPUInitDoesNotFireNoncurrent(t *testing.T) { init := now.AddDate(0, 0, -30) ev := mpuInitEvent("bk", "u1", "logs/foo.txt", init.Unix(), init.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected exactly 1 match (ABORT_MPU only), got %v", matches) } @@ -390,7 +392,7 @@ func TestRouteRegularObjectUnderDualRuleSkipsAbortMPU(t *testing.T) { old := now.AddDate(0, 0, -2) // past the 1d expiration ev := eventCreate("bk", "obj", old.Unix(), 1, old.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected exactly 1 match (EXPIRATION_DAYS only), got %v", matches) } @@ -427,7 +429,7 @@ func TestRouteVersionedNoncurrentEventDoesNotFireFromRouter(t *testing.T) { s3_constants.ExtVersionIdKey: []byte("v1"), } - if got := Route(snap, ev, now); len(got) != 0 { + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 { t.Fatalf("router must not emit noncurrent matches yet, got %v", got) } } @@ -443,7 +445,7 @@ func TestRouteVersionedCurrentEventStaysLatest(t *testing.T) { old := now.AddDate(0, 0, -2) ev := eventCreate("bk", "logs/foo", old.Unix(), 1, old.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match (EXPIRATION_DAYS), got %v", matches) } @@ -467,7 +469,7 @@ func TestRouteNonVersionedBucketIgnoresVersionsSuffix(t *testing.T) { old := now.AddDate(0, 0, -2) ev := eventCreate("bk", "logs/foo.versions/v1", old.Unix(), 1, old.UnixNano()) - matches := Route(snap, ev, now) + matches := Route(context.Background(), snap, ev, now, nil) if len(matches) != 1 { t.Fatalf("expected 1 match, got %v", matches) } @@ -479,11 +481,49 @@ func TestRouteNonVersionedBucketIgnoresVersionsSuffix(t *testing.T) { } } -func TestRouteVersionedExpiredDeleteMarkerSuppressedWithoutSiblings(t *testing.T) { - // ExpiredObjectDeleteMarker requires NumVersions==1 — the marker is - // the sole-survivor. Without sibling listing the router can't - // confirm that, so the rule must NOT fire just because the latest - // is a delete marker. A future PR adds sibling listing. +// markerEventBytes returns the production shape: a file event under +// .versions/v_, with ExtDeleteMarkerKey="true" and +// ExtVersionIdKey populated. Mirrors createDeleteMarker in +// s3api_object_versioning.go. +func markerEvent(bucket, logicalKey, versionID string, mtimeUnix, mtimeNs int64) *reader.Event { + versionPath := logicalKey + s3_constants.VersionsFolder + "/v_" + versionID + ev := eventCreate(bucket, versionPath, mtimeUnix, 0, mtimeNs) + ev.NewEntry.Extended = map[string][]byte{ + s3_constants.ExtDeleteMarkerKey: []byte("true"), + s3_constants.ExtVersionIdKey: []byte(versionID), + } + return ev +} + +// recordingLister captures Survivors calls. Configure with the exact +// state to return; calls list is appended on each invocation so tests +// can assert whether the lister was consulted at all. +type recordingLister struct { + calls []string + survivors Survivors + err error +} + +func (r *recordingLister) Survivors(_ context.Context, bucket, key string) (Survivors, error) { + r.calls = append(r.calls, bucket+"/"+key) + return r.survivors, r.err +} + +func markerLoneEntry(versionID string, mtimeUnix, mtimeNs int64) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "v_" + versionID, + Attributes: &filer_pb.FuseAttributes{ + Mtime: mtimeUnix, + MtimeNs: int32(mtimeNs - mtimeUnix*int64(1e9)), + }, + Extended: map[string][]byte{ + s3_constants.ExtDeleteMarkerKey: []byte("true"), + s3_constants.ExtVersionIdKey: []byte(versionID), + }, + } +} + +func TestRouteVersionedExpiredDeleteMarkerNilListerSuppresses(t *testing.T) { rule := &s3lifecycle.Rule{ ID: "r", Status: s3lifecycle.StatusEnabled, @@ -493,13 +533,239 @@ func TestRouteVersionedExpiredDeleteMarkerSuppressedWithoutSiblings(t *testing.T now := time.Now() old := now.AddDate(0, 0, -1) - ev := eventCreate("bk", "logs/gone", old.Unix(), 0, old.UnixNano()) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 { + t.Fatalf("nil lister must suppress, got %v", got) + } +} + +func TestRouteVersionedExpiredDeleteMarkerSoleSurvivorFires(t *testing.T) { + // Exactly one entry under .versions// — the marker — and no + // bare null version: EXP_DM fires with the LOGICAL key + version_id. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + lister := &recordingLister{survivors: Survivors{ + Count: 1, + LoneEntry: markerLoneEntry("2026-05-09-abc", old.Unix(), old.UnixNano()), + }} + matches := Route(context.Background(), snap, ev, now, lister) + if len(matches) != 1 { + t.Fatalf("expected 1 match (ExpiredDeleteMarker), got %v", matches) + } + m := matches[0] + if m.Result.Action != s3lifecycle.ActionExpireDeleteMarker { + t.Fatalf("Action=%v, want ExpireDeleteMarker", m.Result.Action) + } + if m.ObjectKey != "logs/gone" { + t.Fatalf("ObjectKey=%q, want logical key logs/gone", m.ObjectKey) + } + if m.VersionID != "2026-05-09-abc" { + t.Fatalf("VersionID=%q, want 2026-05-09-abc", m.VersionID) + } + if len(lister.calls) != 1 || lister.calls[0] != "bk/logs/gone" { + t.Fatalf("lister calls=%v, want [bk/logs/gone]", lister.calls) + } +} + +func TestRouteVersionedExpiredDeleteMarkerSiblingsRemainSuppressed(t *testing.T) { + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + lister := &recordingLister{survivors: Survivors{Count: 2}} + if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 { + t.Fatalf("siblings present, must not fire, got %v", got) + } +} + +func TestRouteVersionedExpiredDeleteMarkerNullVersionSuppresses(t *testing.T) { + // A pre-versioning bare-key object (HasNullVersion=true) still survives + // outside .versions/. Firing EXP_DM would re-expose it; suppress. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + lister := &recordingLister{survivors: Survivors{ + Count: 1, + LoneEntry: markerLoneEntry("2026-05-09-abc", old.Unix(), old.UnixNano()), + HasNullVersion: true, + }} + if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 { + t.Fatalf("null-version present, must not fire, got %v", got) + } +} + +func TestRouteVersionedExpiredDeleteMarkerHardDeleteLeavesLoneMarkerFires(t *testing.T) { + // Sequence: object had v1 + DM, hard-delete of v1 leaves DM as the + // sole survivor. The hard-delete event has NewEntry=nil; the router + // must still consult the lister, see the lone DM, and emit a match + // using the LoneEntry's version_id and identity. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + versionPath := "logs/gone" + s3_constants.VersionsFolder + "/v_2026-05-08-old" + ev := &reader.Event{ + TsNs: now.UnixNano(), + Bucket: "bk", + Key: versionPath, + OldEntry: &filer_pb.Entry{Name: "v_2026-05-08-old"}, + } + lister := &recordingLister{survivors: Survivors{ + Count: 1, + LoneEntry: markerLoneEntry("2026-05-09-abc", old.Unix(), old.UnixNano()), + }} + matches := Route(context.Background(), snap, ev, now, lister) + if len(matches) != 1 { + t.Fatalf("expected 1 match after hard-delete, got %v", matches) + } + if matches[0].VersionID != "2026-05-09-abc" { + t.Fatalf("VersionID=%q, want lone-entry's version 2026-05-09-abc", matches[0].VersionID) + } + if matches[0].ObjectKey != "logs/gone" { + t.Fatalf("ObjectKey=%q, want logs/gone", matches[0].ObjectKey) + } +} + +func TestRouteVersionedExpiredDeleteMarkerHardDeleteLoneNonMarkerNoFire(t *testing.T) { + // After a hard-delete the surviving entry is a regular version, not + // a marker. Nothing to expire. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + versionPath := "logs/gone" + s3_constants.VersionsFolder + "/v_2026-05-08-old" + ev := &reader.Event{ + TsNs: now.UnixNano(), + Bucket: "bk", + Key: versionPath, + OldEntry: &filer_pb.Entry{Name: "v_2026-05-08-old"}, + } + regular := &filer_pb.Entry{ + Name: "v_v1", + Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix()}, + Extended: map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}, + } + lister := &recordingLister{survivors: Survivors{Count: 1, LoneEntry: regular}} + if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 { + t.Fatalf("non-marker survivor must not fire, got %v", got) + } +} + +func TestRouteVersionedExpiredDeleteMarkerListerErrorSuppressed(t *testing.T) { + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + lister := &recordingLister{err: errors.New("filer down")} + if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 { + t.Fatalf("lister error must suppress, got %v", got) + } +} + +func TestRouteVersionedExpiredDeleteMarkerInactiveActionSkipsListing(t *testing.T) { + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + // PriorStates omitted => BootstrapComplete=false => action stays inactive. + snap := compileWithVersioned(rule, nil) + + now := time.Now() + old := now.AddDate(0, 0, -1) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + lister := &recordingLister{survivors: Survivors{Count: 1}} + if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 { + t.Fatalf("inactive action must not produce a match, got %v", got) + } + if len(lister.calls) != 0 { + t.Fatalf("inactive action must not consult lister, calls=%v", lister.calls) + } +} + +func TestRouteVersionedRegularVersionCreateSkipsListing(t *testing.T) { + // A non-marker version create under .versions// can never be the + // sole survivor (Count >= 2 by definition), so the lister must NOT + // be consulted on every regular versioned PUT. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpiredObjectDeleteMarker: true, + } + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + versionPath := "logs/keep" + s3_constants.VersionsFolder + "/v_v1" + ev := eventCreate("bk", versionPath, old.Unix(), 100, old.UnixNano()) ev.NewEntry.Extended = map[string][]byte{ - s3_constants.ExtDeleteMarkerKey: {1}, + s3_constants.ExtVersionIdKey: []byte("v1"), } - if got := Route(snap, ev, now); len(got) != 0 { - t.Fatalf("ExpiredDeleteMarker without sibling count must not fire, got %v", got) + lister := &recordingLister{survivors: Survivors{Count: 1}} + if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 { + t.Fatalf("regular version create must not fire EXP_DM, got %v", got) + } + if len(lister.calls) != 0 { + t.Fatalf("lister consulted for regular version create: calls=%v", lister.calls) + } +} + +func TestRouteVersionedDeleteMarkerNoExpDMRuleSkipsListing(t *testing.T) { + rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1} + snap := compileWithVersioned(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -1) + ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano()) + + lister := &recordingLister{survivors: Survivors{Count: 1}} + Route(context.Background(), snap, ev, now, lister) + if len(lister.calls) != 0 { + t.Fatalf("lister consulted without EXP_DM rule: calls=%v", lister.calls) } } @@ -555,7 +821,7 @@ func TestRouteVersionedAllVersionFolderPathsSkipped(t *testing.T) { if tc.isDir { ev.NewEntry.IsDirectory = true } - if got := Route(snap, ev, now); len(got) != 0 { + if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 { t.Fatalf("version-folder event should be skipped, got %v", got) } })