diff --git a/weed/s3api/s3api_lifecycle_canonical.go b/weed/s3api/s3api_lifecycle_canonical.go new file mode 100644 index 000000000..a0e4ac5a6 --- /dev/null +++ b/weed/s3api/s3api_lifecycle_canonical.go @@ -0,0 +1,101 @@ +package s3api + +import ( + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// LifecycleToCanonical converts the XML-deserialized Lifecycle into the flat +// s3lifecycle.Rule shape the engine compiles against. One XML maps to +// exactly one s3lifecycle.Rule (with potentially multiple action sub-fields +// populated); the engine then expands each into its compiled actions via +// s3lifecycle.RuleActionKinds. +// +// Filter resolution mirrors AWS semantics: the optional element may +// contain a single | | , or be absent (in which case the +// older top-level applies). When is used, its sub-elements +// (Prefix + multiple Tags + size filters) are flattened into the canonical +// Rule's individual fields. +func LifecycleToCanonical(lc *Lifecycle) []*s3lifecycle.Rule { + if lc == nil { + return nil + } + out := make([]*s3lifecycle.Rule, 0, len(lc.Rules)) + for i := range lc.Rules { + out = append(out, ruleToCanonical(&lc.Rules[i])) + } + return out +} + +func ruleToCanonical(r *Rule) *s3lifecycle.Rule { + out := &s3lifecycle.Rule{ + ID: r.ID, + Status: string(r.Status), + } + + // Prefix: or or top-level . + prefix, tags, sizeGT, sizeLT := flattenFilter(&r.Filter) + if prefix == "" && r.Prefix.set { + prefix = r.Prefix.val + } + out.Prefix = prefix + if len(tags) > 0 { + out.FilterTags = tags + } + out.FilterSizeGreaterThan = sizeGT + out.FilterSizeLessThan = sizeLT + + // Expiration sub-fields. + if r.Expiration.set { + out.ExpirationDays = r.Expiration.Days + if !r.Expiration.Date.Time.IsZero() { + out.ExpirationDate = r.Expiration.Date.Time + } + if r.Expiration.DeleteMarker.set { + out.ExpiredObjectDeleteMarker = r.Expiration.DeleteMarker.val + } + } + + // Non-current version expiration. + if r.NoncurrentVersionExpiration.set { + out.NoncurrentVersionExpirationDays = r.NoncurrentVersionExpiration.NoncurrentDays + out.NewerNoncurrentVersions = r.NoncurrentVersionExpiration.NewerNoncurrentVersions + } + + // Abort multipart. + if r.AbortIncompleteMultipartUpload.set { + out.AbortMPUDaysAfterInitiation = r.AbortIncompleteMultipartUpload.DaysAfterInitiation + } + + return out +} + +// flattenFilter pulls Prefix / Tags / Size constraints out of the XML Filter +// element. Returns zero values when the field is unset; the caller falls back +// to the rule's top-level Prefix when prefix is "". +func flattenFilter(f *Filter) (prefix string, tags map[string]string, sizeGT, sizeLT int64) { + if !f.set { + return + } + if f.andSet { + if f.And.Prefix.set { + prefix = f.And.Prefix.val + } + if len(f.And.Tags) > 0 { + tags = make(map[string]string, len(f.And.Tags)) + for _, t := range f.And.Tags { + tags[t.Key] = t.Value + } + } + sizeGT = f.And.ObjectSizeGreaterThan + sizeLT = f.And.ObjectSizeLessThan + return + } + if f.tagSet { + tags = map[string]string{f.Tag.Key: f.Tag.Value} + } else if f.Prefix.set { + prefix = f.Prefix.val + } + sizeGT = f.ObjectSizeGreaterThan + sizeLT = f.ObjectSizeLessThan + return +} diff --git a/weed/s3api/s3api_lifecycle_canonical_test.go b/weed/s3api/s3api_lifecycle_canonical_test.go new file mode 100644 index 000000000..543f72861 --- /dev/null +++ b/weed/s3api/s3api_lifecycle_canonical_test.go @@ -0,0 +1,196 @@ +package s3api + +import ( + "encoding/xml" + "reflect" + "testing" + "time" +) + +// parseLifecycle is a thin helper for tests; production code reads XML via the +// regular bucket-config decoder, this shortcut keeps the test focused on the +// canonical conversion. +func parseLifecycle(t *testing.T, xmlSrc string) *Lifecycle { + t.Helper() + lc := &Lifecycle{} + if err := xml.Unmarshal([]byte(xmlSrc), lc); err != nil { + t.Fatalf("unmarshal: %v", err) + } + return lc +} + +func TestLifecycleToCanonical_TopLevelPrefix(t *testing.T) { + lc := parseLifecycle(t, ` + + + r1 + Enabled + logs/ + 30 + +`) + got := LifecycleToCanonical(lc) + if len(got) != 1 { + t.Fatalf("want 1 rule, got %d", len(got)) + } + r := got[0] + if r.ID != "r1" || r.Status != "Enabled" || r.Prefix != "logs/" || r.ExpirationDays != 30 { + t.Fatalf("unexpected: %+v", r) + } +} + +func TestLifecycleToCanonical_FilterPrefix(t *testing.T) { + lc := parseLifecycle(t, ` + + + Enabled + logs/ + 30 + +`) + got := LifecycleToCanonical(lc) + if got[0].Prefix != "logs/" { + t.Fatalf("want logs/, got %q", got[0].Prefix) + } +} + +func TestLifecycleToCanonical_FilterTagAndSize(t *testing.T) { + lc := parseLifecycle(t, ` + + + Enabled + + + data/ + envprod + tiercold + 1024 + 10485760 + + + 90 + +`) + got := LifecycleToCanonical(lc)[0] + if got.Prefix != "data/" { + t.Fatalf("prefix want data/, got %q", got.Prefix) + } + wantTags := map[string]string{"env": "prod", "tier": "cold"} + if !reflect.DeepEqual(got.FilterTags, wantTags) { + t.Fatalf("tags want %v, got %v", wantTags, got.FilterTags) + } + if got.FilterSizeGreaterThan != 1024 || got.FilterSizeLessThan != 10485760 { + t.Fatalf("size: gt=%d lt=%d", got.FilterSizeGreaterThan, got.FilterSizeLessThan) + } +} + +func TestLifecycleToCanonical_SingleTagFilter(t *testing.T) { + lc := parseLifecycle(t, ` + + + Enabled + + envprod + + 30 + +`) + got := LifecycleToCanonical(lc)[0] + if !reflect.DeepEqual(got.FilterTags, map[string]string{"env": "prod"}) { + t.Fatalf("tags want {env:prod}, got %v", got.FilterTags) + } +} + +func TestLifecycleToCanonical_MultipleActions(t *testing.T) { + // One XML with three concurrent actions: Expiration.Days, + // NoncurrentVersionExpiration, AbortMultipartUpload. All must populate + // the corresponding canonical fields so RuleActionKinds expands them + // into three compiled actions downstream. + lc := parseLifecycle(t, ` + + + multi + Enabled + data/ + 90 + + 30 + 3 + + + 7 + + +`) + got := LifecycleToCanonical(lc)[0] + if got.ExpirationDays != 90 { + t.Fatalf("ExpirationDays want 90, got %d", got.ExpirationDays) + } + if got.NoncurrentVersionExpirationDays != 30 || got.NewerNoncurrentVersions != 3 { + t.Fatalf("NoncurrentVersion: %+v", got) + } + if got.AbortMPUDaysAfterInitiation != 7 { + t.Fatalf("AbortMPU want 7, got %d", got.AbortMPUDaysAfterInitiation) + } +} + +func TestLifecycleToCanonical_ExpirationDate(t *testing.T) { + lc := parseLifecycle(t, ` + + + Enabled + + 2025-06-15T00:00:00Z + +`) + got := LifecycleToCanonical(lc)[0] + want, _ := time.Parse(time.RFC3339, "2025-06-15T00:00:00Z") + if !got.ExpirationDate.Equal(want) { + t.Fatalf("date want %v, got %v", want, got.ExpirationDate) + } +} + +func TestLifecycleToCanonical_ExpiredObjectDeleteMarker(t *testing.T) { + lc := parseLifecycle(t, ` + + + Enabled + + true + +`) + got := LifecycleToCanonical(lc)[0] + if !got.ExpiredObjectDeleteMarker { + t.Fatalf("want ExpiredObjectDeleteMarker=true") + } +} + +func TestLifecycleToCanonical_DisabledRulePreserved(t *testing.T) { + // The engine's mode gate decides scheduling; conversion must preserve + // the Status verbatim regardless. + lc := parseLifecycle(t, ` + + + Disabled + x/ + 30 + +`) + got := LifecycleToCanonical(lc)[0] + if got.Status != "Disabled" { + t.Fatalf("status want Disabled, got %q", got.Status) + } +} + +func TestLifecycleToCanonical_NilSafe(t *testing.T) { + if got := LifecycleToCanonical(nil); got != nil { + t.Fatalf("nil lc should return nil, got %v", got) + } +} + +func TestLifecycleToCanonical_EmptyRules(t *testing.T) { + got := LifecycleToCanonical(&Lifecycle{}) + if len(got) != 0 { + t.Fatalf("empty rules should be empty slice, got %d", len(got)) + } +} diff --git a/weed/s3api/s3lifecycle/action_kind.go b/weed/s3api/s3lifecycle/action_kind.go index 8893149a2..4279fb476 100644 --- a/weed/s3api/s3lifecycle/action_kind.go +++ b/weed/s3api/s3lifecycle/action_kind.go @@ -1,22 +1,27 @@ package s3lifecycle -// ActionKind identifies a single compiled lifecycle action under one XML -// rule. A single XML may declare multiple action sub-elements in -// parallel, each yielding a separate compiled action with its own delay -// group, mode, pending stream, and durable state directory. -// -// The values here mirror the wire-form ActionKind enum in -// weed/pb/s3_lifecycle.proto (offset by the UNSPECIFIED sentinel at 0). +// ActionKey is the engine-wide identity of one compiled lifecycle action. +// Bucket is part of the key because two buckets may carry rules with +// identical RuleHash; without scoping they'd collide in any keyed map. +// The on-disk path /etc/s3/lifecycle//// +// mirrors this shape. +type ActionKey struct { + Bucket string + RuleHash [8]byte + ActionKind ActionKind +} + +// ActionKind values mirror the wire-form enum in s3_lifecycle.proto. type ActionKind int const ( - ActionKindUnspecified ActionKind = iota // matches proto ACTION_KIND_UNSPECIFIED - ActionKindExpirationDays // Expiration.Days - ActionKindExpirationDate // Expiration.Date - ActionKindNoncurrentDays // NoncurrentVersionExpiration.NoncurrentDays (with optional NewerNoncurrent retention) - ActionKindNewerNoncurrent // NoncurrentVersionExpiration.NewerNoncurrentVersions (count-only, no NoncurrentDays) - ActionKindAbortMPU // AbortIncompleteMultipartUpload.DaysAfterInitiation - ActionKindExpiredDeleteMarker // Expiration.ExpiredObjectDeleteMarker + ActionKindUnspecified ActionKind = iota + ActionKindExpirationDays + ActionKindExpirationDate + ActionKindNoncurrentDays + ActionKindNewerNoncurrent + ActionKindAbortMPU + ActionKindExpiredDeleteMarker ) // String returns the leaf-directory name used in @@ -40,16 +45,10 @@ func (k ActionKind) String() string { } } -// RuleActionKinds returns the compiled actions a single XML rule expands to. -// Empty when no action sub-element is populated. Order is deterministic so -// callers can hash / iterate stably: -// -// EXPIRATION_DAYS, EXPIRATION_DATE, EXPIRED_DELETE_MARKER, -// NONCURRENT_DAYS, NEWER_NONCURRENT, ABORT_MPU -// -// Note: NewerNoncurrentVersions is paired with NoncurrentDays into a single -// NONCURRENT_DAYS action when both are set; only when NewerNoncurrent is set -// alone (no day threshold) does it produce a NEWER_NONCURRENT action. +// RuleActionKinds returns the compiled actions a single XML rule expands to, +// in deterministic order. NewerNoncurrentVersions paired with NoncurrentDays +// is subsumed into NONCURRENT_DAYS; only stand-alone NewerNoncurrent +// produces a NEWER_NONCURRENT action. func RuleActionKinds(rule *Rule) []ActionKind { if rule == nil { return nil diff --git a/weed/s3api/s3lifecycle/due_at.go b/weed/s3api/s3lifecycle/due_at.go index b8aa1abcd..67d4a2a53 100644 --- a/weed/s3api/s3lifecycle/due_at.go +++ b/weed/s3api/s3lifecycle/due_at.go @@ -2,15 +2,9 @@ package s3lifecycle import "time" -// ComputeDueAt returns the earliest wall-clock time the (rule, kind) compiled -// action can fire for info given the object's current shape. Returns the -// zero time when the action cannot fire for this entry (filter rejects, kind -// not declared on the rule, wrong object shape, etc.). -// -// Used by the reader/bootstrap to decide pending-vs-inline-delete for one -// specific action. Sibling actions of the same XML rule are computed -// separately so a rule's 7d AbortMPU due time does not influence its 90d -// ExpirationDays sibling. +// ComputeDueAt returns the earliest wall-clock time the (rule, kind) action +// can fire for info. Returns zero time when no action can fire for this +// entry. Used by reader/bootstrap to decide pending vs. inline-delete. func ComputeDueAt(rule *Rule, kind ActionKind, info *ObjectInfo) time.Time { if rule == nil || info == nil || rule.Status != StatusEnabled { return time.Time{} @@ -45,7 +39,6 @@ func ComputeDueAt(rule *Rule, kind ActionKind, info *ObjectInfo) time.Time { return base.AddDate(0, 0, rule.NoncurrentVersionExpirationDays) } case ActionKindNewerNoncurrent: - // Pure count-based: only when NoncurrentDays is unset. if !info.IsLatest && rule.NoncurrentVersionExpirationDays == 0 && rule.NewerNoncurrentVersions > 0 { if !info.SuccessorModTime.IsZero() { return info.SuccessorModTime diff --git a/weed/s3api/s3lifecycle/engine/compile.go b/weed/s3api/s3lifecycle/engine/compile.go new file mode 100644 index 000000000..45fe40bb4 --- /dev/null +++ b/weed/s3api/s3lifecycle/engine/compile.go @@ -0,0 +1,126 @@ +package engine + +import ( + "bytes" + "sort" + "sync/atomic" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +var snapshotIDSeq atomic.Uint64 + +// CompileInput: at most one entry per Bucket. Duplicates would overwrite +// earlier entries' BucketIndex. +type CompileInput struct { + Bucket string + Rules []*s3lifecycle.Rule + Versioned bool +} + +// PriorState carries durable per-action state into Compile. Missing keys +// are treated as bootstrap_complete=false. +type PriorState struct { + BootstrapComplete bool + Mode RuleMode +} + +// MetaLogRetention=0 means unbounded; the retention gate doesn't trip. +type CompileOptions struct { + MetaLogRetention time.Duration + BootstrapLookbackMin time.Duration + PriorStates map[s3lifecycle.ActionKey]PriorState +} + +const defaultBootstrapLookbackMin = 5 * s3lifecycle.SmallDelay + +func (e *Engine) Compile(inputs []CompileInput, opts CompileOptions) *Snapshot { + if opts.BootstrapLookbackMin == 0 { + opts.BootstrapLookbackMin = defaultBootstrapLookbackMin + } + + snap := &Snapshot{ + id: snapshotIDSeq.Add(1), + buckets: make(map[string]*BucketIndex), + actions: make(map[s3lifecycle.ActionKey]*CompiledAction), + originalDelayGroups: make(map[time.Duration][]s3lifecycle.ActionKey), + dateActions: make(map[s3lifecycle.ActionKey]time.Time), + } + + for _, in := range inputs { + bi := &BucketIndex{bucket: in.Bucket, versioned: in.Versioned} + snap.buckets[in.Bucket] = bi + + for _, rule := range in.Rules { + ruleHash := s3lifecycle.RuleHash(rule) + for _, kind := range s3lifecycle.RuleActionKinds(rule) { + key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: ruleHash, ActionKind: kind} + // Durable Mode wins; decideMode is only the seed for new + // or legacy-zero actions. Otherwise lag-fallback / operator + // SCAN_ONLY would re-promote on every rebuild. + prior, hasPrior := opts.PriorStates[key] + mode := prior.Mode + if !hasPrior || mode == ModeUnspecified { + mode = decideMode(rule, kind, opts.MetaLogRetention, opts.BootstrapLookbackMin) + } + active := prior.BootstrapComplete && mode == ModeEventDriven + + ca := &CompiledAction{ + Rule: rule, + Bucket: in.Bucket, + Key: key, + Delay: s3lifecycle.MinTriggerAge(rule, kind), + PredicateSensitive: rulePredicateSensitive(rule), + Mode: mode, + } + if active { + ca.markActive() + } + snap.actions[key] = ca + bi.actionKeys = append(bi.actionKeys, key) + + // Index every action regardless of `active`; routing + // re-filters on IsActive() so MarkActive flips are visible + // without a recompile. + if mode == ModeScanAtDate { + snap.dateActions[key] = rule.ExpirationDate + } + if mode == ModeEventDriven { + snap.originalDelayGroups[ca.Delay] = append(snap.originalDelayGroups[ca.Delay], key) + if ca.PredicateSensitive { + snap.predicateActions = append(snap.predicateActions, key) + } + } + } + } + } + + snap.allActionsSorted = make([]*CompiledAction, 0, len(snap.actions)) + for _, a := range snap.actions { + snap.allActionsSorted = append(snap.allActionsSorted, a) + } + sort.Slice(snap.allActionsSorted, func(i, j int) bool { + a, b := snap.allActionsSorted[i], snap.allActionsSorted[j] + if a.Bucket != b.Bucket { + return a.Bucket < b.Bucket + } + if c := bytes.Compare(a.Key.RuleHash[:], b.Key.RuleHash[:]); c != 0 { + return c < 0 + } + return a.Key.ActionKind < b.Key.ActionKind + }) + + e.current.Store(snap) + return snap +} + +// rulePredicateSensitive: only tag filters can flip post-PUT. Size is +// immutable once written; a size change is a fresh write through the +// original-write stream. +func rulePredicateSensitive(rule *s3lifecycle.Rule) bool { + if rule == nil { + return false + } + return len(rule.FilterTags) > 0 +} diff --git a/weed/s3api/s3lifecycle/engine/engine.go b/weed/s3api/s3lifecycle/engine/engine.go new file mode 100644 index 000000000..7ac0b8856 --- /dev/null +++ b/weed/s3api/s3lifecycle/engine/engine.go @@ -0,0 +1,120 @@ +// Package engine compiles per-bucket lifecycle rules into a CompiledAction +// snapshot. One XML expands into N compiled actions (one per populated +// action sub-element); each is keyed by ActionKey{bucket, rule_hash, +// action_kind} so sibling actions schedule and degrade independently. +package engine + +import ( + "sync/atomic" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +type Engine struct { + current atomic.Pointer[Snapshot] +} + +func New() *Engine { + e := &Engine{} + e.current.Store(&Snapshot{actions: map[s3lifecycle.ActionKey]*CompiledAction{}}) + return e +} + +func (e *Engine) Snapshot() *Snapshot { return e.current.Load() } + +// Snapshot fields are append-only after Compile except CompiledAction.engineState, +// which transitions inactive -> active atomically via markActive. +type Snapshot struct { + id uint64 + buckets map[string]*BucketIndex + actions map[s3lifecycle.ActionKey]*CompiledAction + + allActionsSorted []*CompiledAction + + // Routing indexes hold every ActionKey by mode regardless of activation; + // dispatch filters on IsActive(). + originalDelayGroups map[time.Duration][]s3lifecycle.ActionKey + predicateActions []s3lifecycle.ActionKey + dateActions map[s3lifecycle.ActionKey]time.Time +} + +func (s *Snapshot) SnapshotID() uint64 { return s.id } + +type BucketIndex struct { + bucket string + versioned bool + actionKeys []s3lifecycle.ActionKey +} + +type CompiledAction struct { + Rule *s3lifecycle.Rule + Bucket string + Key s3lifecycle.ActionKey + Delay time.Duration + PredicateSensitive bool + Mode RuleMode + engineState atomic.Uint32 +} + +const ( + engineStateInactive uint32 = 0 + engineStateActive uint32 = 1 +) + +func (a *CompiledAction) IsActive() bool { + return a.engineState.Load() == engineStateActive +} + +func (a *CompiledAction) markActive() { + a.engineState.Store(engineStateActive) +} + +// MarkActive mirrors the in-memory hint after a durable bootstrap_complete +// write. No-op if the key isn't in the snapshot. +func (s *Snapshot) MarkActive(key s3lifecycle.ActionKey) { + if a, ok := s.actions[key]; ok { + a.markActive() + } +} + +func (s *Snapshot) Action(key s3lifecycle.ActionKey) *CompiledAction { + return s.actions[key] +} + +// AllActions: caller must not mutate. +func (s *Snapshot) AllActions() []*CompiledAction { + return s.allActionsSorted +} + +// OriginalDelayGroups / PredicateActions / DateActions return defensive copies. +func (s *Snapshot) OriginalDelayGroups() map[time.Duration][]s3lifecycle.ActionKey { + out := make(map[time.Duration][]s3lifecycle.ActionKey, len(s.originalDelayGroups)) + for d, keys := range s.originalDelayGroups { + copied := make([]s3lifecycle.ActionKey, len(keys)) + copy(copied, keys) + out[d] = copied + } + return out +} + +func (s *Snapshot) PredicateActions() []s3lifecycle.ActionKey { + out := make([]s3lifecycle.ActionKey, len(s.predicateActions)) + copy(out, s.predicateActions) + return out +} + +func (s *Snapshot) DateActions() map[s3lifecycle.ActionKey]time.Time { + out := make(map[s3lifecycle.ActionKey]time.Time, len(s.dateActions)) + for k, v := range s.dateActions { + out[k] = v + } + return out +} + +func (s *Snapshot) BucketVersioned(bucket string) bool { + if bi, ok := s.buckets[bucket]; ok { + return bi.versioned + } + return false +} diff --git a/weed/s3api/s3lifecycle/engine/engine_test.go b/weed/s3api/s3lifecycle/engine/engine_test.go new file mode 100644 index 000000000..cf28cc1e5 --- /dev/null +++ b/weed/s3api/s3lifecycle/engine/engine_test.go @@ -0,0 +1,310 @@ +package engine + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +func ruleExpDays(id, prefix string, days int) *s3lifecycle.Rule { + return &s3lifecycle.Rule{ + ID: id, + Status: s3lifecycle.StatusEnabled, + Prefix: prefix, + ExpirationDays: days, + } +} + +func TestCompile_SingleRuleSingleAction(t *testing.T) { + e := New() + rule := ruleExpDays("r1", "logs/", 30) + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{ + PriorStates: map[s3lifecycle.ActionKey]PriorState{ + {Bucket: "b1", RuleHash: s3lifecycle.RuleHash(rule), ActionKind: s3lifecycle.ActionKindExpirationDays}: { + BootstrapComplete: true, + }, + }, + }) + if got := len(snap.actions); got != 1 { + t.Fatalf("want 1 action, got %d", got) + } + for _, a := range snap.actions { + if !a.IsActive() { + t.Fatalf("bootstrap_complete + EVENT_DRIVEN should activate, got mode=%v", a.Mode) + } + if a.Mode != ModeEventDriven { + t.Fatalf("want EVENT_DRIVEN, got %v", a.Mode) + } + if a.Delay != 30*24*time.Hour { + t.Fatalf("want 30d delay, got %v", a.Delay) + } + } +} + +func TestCompile_MultiAction_SiblingsHaveOwnEntries(t *testing.T) { + // One XML rule with three actions -> three CompiledActions, three + // distinct ActionKeys, three independent delay group memberships. + rule := &s3lifecycle.Rule{ + ID: "multi", + Status: s3lifecycle.StatusEnabled, + Prefix: "data/", + ExpirationDays: 90, + NoncurrentVersionExpirationDays: 30, + AbortMPUDaysAfterInitiation: 7, + } + rh := s3lifecycle.RuleHash(rule) + prior := map[s3lifecycle.ActionKey]PriorState{} + for _, k := range s3lifecycle.RuleActionKinds(rule) { + prior[s3lifecycle.ActionKey{Bucket: "b1", RuleHash: rh, ActionKind: k}] = PriorState{BootstrapComplete: true} + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: prior}) + + if got := len(snap.actions); got != 3 { + t.Fatalf("want 3 actions, got %d", got) + } + wantDelays := map[time.Duration]bool{ + 90 * 24 * time.Hour: true, // ExpirationDays + 30 * 24 * time.Hour: true, // NoncurrentDays + 7 * 24 * time.Hour: true, // AbortMPU + } + for delay, keys := range snap.originalDelayGroups { + if !wantDelays[delay] { + t.Fatalf("unexpected delay group %v", delay) + } + if len(keys) != 1 { + t.Fatalf("delay %v should hold exactly its own action, got %d", delay, len(keys)) + } + } + if len(snap.originalDelayGroups) != 3 { + t.Fatalf("want 3 delay groups, got %d", len(snap.originalDelayGroups)) + } +} + +func TestCompile_BootstrapPendingIndexedButInactive(t *testing.T) { + // Without prior bootstrap_complete=true the action is pending_bootstrap. + // It IS indexed in originalDelayGroups (so a later MarkActive flip is + // routable without recompile), but IsActive() reads false so the + // reader's IsActive-filter skips dispatch. + rule := ruleExpDays("r1", "logs/", 30) + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{}) + + for _, a := range snap.actions { + if a.IsActive() { + t.Fatalf("pending_bootstrap action should not be active") + } + } + if len(snap.originalDelayGroups) != 1 { + t.Fatalf("EVENT_DRIVEN action should be indexed even when inactive, got %v", snap.originalDelayGroups) + } +} + +func TestCompile_RetentionGate(t *testing.T) { + // A 90d ExpirationDays rule with 30d retention should land in scan_only. + // A 7d rule (under retention - lookback) stays event_driven. + long := ruleExpDays("long", "x/", 90) + short := ruleExpDays("short", "y/", 1) + rules := []*s3lifecycle.Rule{long, short} + prior := map[s3lifecycle.ActionKey]PriorState{} + for _, r := range rules { + k := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(r), ActionKind: s3lifecycle.ActionKindExpirationDays} + prior[k] = PriorState{BootstrapComplete: true} + } + + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: rules}}, CompileOptions{ + MetaLogRetention: 30 * 24 * time.Hour, + BootstrapLookbackMin: 5 * time.Minute, + PriorStates: prior, + }) + + for _, a := range snap.actions { + if a.Rule.ID == "long" && a.Mode != ModeScanOnly { + t.Fatalf("90d rule under 30d retention should be scan_only, got %v", a.Mode) + } + if a.Rule.ID == "short" && a.Mode != ModeEventDriven { + t.Fatalf("1d rule should be event_driven, got %v", a.Mode) + } + } +} + +func TestCompile_RetentionUnboundedNeverGates(t *testing.T) { + // MetaLogRetention=0 means unbounded (default deployment); even a 100y + // rule should stay event_driven. + rule := ruleExpDays("decade", "x/", 36500) + prior := map[s3lifecycle.ActionKey]PriorState{ + {RuleHash: s3lifecycle.RuleHash(rule), ActionKind: s3lifecycle.ActionKindExpirationDays}: {BootstrapComplete: true}, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: prior}) + for _, a := range snap.actions { + if a.Mode != ModeEventDriven { + t.Fatalf("unbounded retention should keep event_driven, got %v", a.Mode) + } + } +} + +func TestCompile_SiblingsDegradeIndependently(t *testing.T) { + // 90d ExpirationDays + 7d AbortMPU under 30d retention: ExpirationDays + // degrades to scan_only, AbortMPU stays event_driven. The whole point + // of per-action keying. + rule := &s3lifecycle.Rule{ + ID: "mixed", + Status: s3lifecycle.StatusEnabled, + Prefix: "x/", + ExpirationDays: 90, + AbortMPUDaysAfterInitiation: 7, + } + rh := s3lifecycle.RuleHash(rule) + prior := map[s3lifecycle.ActionKey]PriorState{ + {RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays}: {BootstrapComplete: true}, + {RuleHash: rh, ActionKind: s3lifecycle.ActionKindAbortMPU}: {BootstrapComplete: true}, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{ + MetaLogRetention: 30 * 24 * time.Hour, + BootstrapLookbackMin: 5 * time.Minute, + PriorStates: prior, + }) + expDaysKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays} + mpuKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: rh, ActionKind: s3lifecycle.ActionKindAbortMPU} + if snap.actions[expDaysKey].Mode != ModeScanOnly { + t.Fatalf("ExpirationDays should be scan_only under 30d retention") + } + if snap.actions[mpuKey].Mode != ModeEventDriven { + t.Fatalf("AbortMPU should stay event_driven (sibling degrades independently)") + } +} + +func TestCompile_PriorModePreservedOverDecideMode(t *testing.T) { + // A durably-persisted SCAN_ONLY (or DISABLED, or any degraded mode) + // must not be re-promoted to EVENT_DRIVEN by decideMode on every + // Compile. Otherwise lag-fallback / operator pause / manual scan-only + // don't survive an engine rebuild. + rule := ruleExpDays("r", "x/", 30) + rh := s3lifecycle.RuleHash(rule) + key := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays} + + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{ + PriorStates: map[s3lifecycle.ActionKey]PriorState{ + key: {BootstrapComplete: true, Mode: ModeScanOnly}, + }, + }) + if snap.actions[key].Mode != ModeScanOnly { + t.Fatalf("durable Mode=ScanOnly should win over decideMode, got %v", snap.actions[key].Mode) + } + if snap.actions[key].IsActive() { + t.Fatalf("ScanOnly action must not be active") + } + // And: a missing PriorState falls through to decideMode as before. + rule2 := ruleExpDays("fresh", "x/", 30) + key2 := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(rule2), ActionKind: s3lifecycle.ActionKindExpirationDays} + snap2 := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule2}}}, CompileOptions{}) + if snap2.actions[key2].Mode != ModeEventDriven { + t.Fatalf("missing prior should fall through to decideMode (EventDriven), got %v", snap2.actions[key2].Mode) + } +} + +func TestCompile_ExpirationDateScansAtDate(t *testing.T) { + date := time.Date(2025, 6, 15, 0, 0, 0, 0, time.UTC) + rule := &s3lifecycle.Rule{ + ID: "d", + Status: s3lifecycle.StatusEnabled, + Prefix: "x/", + ExpirationDate: date, + } + prior := map[s3lifecycle.ActionKey]PriorState{ + {RuleHash: s3lifecycle.RuleHash(rule), ActionKind: s3lifecycle.ActionKindExpirationDate}: {BootstrapComplete: true}, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: prior}) + if len(snap.dateActions) != 1 { + t.Fatalf("want 1 date action, got %d", len(snap.dateActions)) + } + for _, d := range snap.dateActions { + if !d.Equal(date) { + t.Fatalf("date want %v, got %v", date, d) + } + } +} + +func TestCompile_DisabledRuleNeverActivates(t *testing.T) { + rule := ruleExpDays("d", "x/", 30) + rule.Status = s3lifecycle.StatusDisabled + prior := map[s3lifecycle.ActionKey]PriorState{ + {RuleHash: s3lifecycle.RuleHash(rule), ActionKind: s3lifecycle.ActionKindExpirationDays}: {BootstrapComplete: true}, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: prior}) + for _, a := range snap.actions { + if a.Mode != ModeDisabled || a.IsActive() { + t.Fatalf("disabled rule must be ModeDisabled and inactive") + } + } +} + +func TestSnapshot_MarkActiveFlipsRoutingFilter(t *testing.T) { + rule := ruleExpDays("r", "x/", 30) + e := New() + snap := e.Compile([]CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{}) + key := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(rule), ActionKind: s3lifecycle.ActionKindExpirationDays} + + if snap.actions[key].IsActive() { + t.Fatalf("should start inactive") + } + snap.MarkActive(key) + if !snap.actions[key].IsActive() { + t.Fatalf("MarkActive should flip the bit") + } + // MarkActive on a missing key is a no-op (stale callback after rebuild). + snap.MarkActive(s3lifecycle.ActionKey{}) +} + +func TestCompile_CrossBucketIdenticalRulesDoNotCollide(t *testing.T) { + // Two buckets carry rules whose XML — and therefore RuleHash — is + // identical. ActionKey must be bucket-scoped so the second bucket's + // CompiledAction does not overwrite the first's in snap.actions. + rule := ruleExpDays("shared", "x/", 30) + rh := s3lifecycle.RuleHash(rule) + prior := map[s3lifecycle.ActionKey]PriorState{ + {Bucket: "alpha", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays}: {BootstrapComplete: true}, + {Bucket: "beta", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays}: {BootstrapComplete: true}, + } + e := New() + snap := e.Compile([]CompileInput{ + {Bucket: "alpha", Rules: []*s3lifecycle.Rule{rule}}, + {Bucket: "beta", Rules: []*s3lifecycle.Rule{rule}}, + }, CompileOptions{PriorStates: prior}) + + if got := len(snap.actions); got != 2 { + t.Fatalf("want 2 actions (one per bucket), got %d", got) + } + alphaKey := s3lifecycle.ActionKey{Bucket: "alpha", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays} + betaKey := s3lifecycle.ActionKey{Bucket: "beta", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays} + if snap.actions[alphaKey] == nil || snap.actions[alphaKey].Bucket != "alpha" { + t.Fatalf("alpha bucket action missing or wrong bucket") + } + if snap.actions[betaKey] == nil || snap.actions[betaKey].Bucket != "beta" { + t.Fatalf("beta bucket action missing or wrong bucket") + } +} + +func TestEngine_SnapshotAtomicSwap(t *testing.T) { + e := New() + r1 := ruleExpDays("r1", "a/", 1) + snap1 := e.Compile([]CompileInput{{Bucket: "b", Rules: []*s3lifecycle.Rule{r1}}}, CompileOptions{}) + if snap1.SnapshotID() == 0 { + t.Fatalf("snapshot id should be > 0") + } + r2 := ruleExpDays("r2", "b/", 2) + snap2 := e.Compile([]CompileInput{{Bucket: "b", Rules: []*s3lifecycle.Rule{r2}}}, CompileOptions{}) + if snap2.SnapshotID() <= snap1.SnapshotID() { + t.Fatalf("snapshot id should be monotonic") + } + if e.Snapshot() != snap2 { + t.Fatalf("Engine.Snapshot should return the latest") + } +} diff --git a/weed/s3api/s3lifecycle/engine/match.go b/weed/s3api/s3lifecycle/engine/match.go new file mode 100644 index 000000000..72205f221 --- /dev/null +++ b/weed/s3api/s3lifecycle/engine/match.go @@ -0,0 +1,134 @@ +package engine + +import ( + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +type EventShape int + +const ( + EventShapeUnknown EventShape = iota + // EventShapeOriginalWrite: any event resetting AWS LastModified + // (fresh PUT, overwrite, MPU init, version flip). + EventShapeOriginalWrite + // EventShapePredicateChange: tags / Extended changed without an mtime + // reset. + EventShapePredicateChange +) + +// Event is the routing-relevant slice of a meta-log event. Kept minimal so +// the engine doesn't depend on filer_pb. +type Event struct { + Shape EventShape + Bucket string + Path string + Tags map[string]string + Size int64 + IsLatest bool + IsDeleteMarker bool + IsMPUInit bool + EventTime time.Time +} + +// MatchOriginalWrite returns active ActionKeys in the given delay group +// whose filter matches the event. +func (s *Snapshot) MatchOriginalWrite(ev *Event, delay time.Duration) []s3lifecycle.ActionKey { + if ev == nil || ev.Shape != EventShapeOriginalWrite { + return nil + } + keys := s.originalDelayGroups[delay] + if len(keys) == 0 { + return nil + } + return s.filterMatching(keys, ev) +} + +// MatchPredicateChange returns active predicate-sensitive ActionKeys whose +// filter matches the event. +func (s *Snapshot) MatchPredicateChange(ev *Event) []s3lifecycle.ActionKey { + if ev == nil || ev.Shape != EventShapePredicateChange { + return nil + } + if len(s.predicateActions) == 0 { + return nil + } + return s.filterMatching(s.predicateActions, ev) +} + +// MatchPath returns active ActionKeys for a specific bucket+path. ev=nil +// applies prefix matching only; pass a non-nil Event to also gate on tags +// and size. +func (s *Snapshot) MatchPath(bucket, path string, ev *Event) []s3lifecycle.ActionKey { + bi := s.buckets[bucket] + if bi == nil { + return nil + } + out := make([]s3lifecycle.ActionKey, 0, len(bi.actionKeys)) + for _, k := range bi.actionKeys { + a := s.actions[k] + if a == nil || !a.IsActive() { + continue + } + if !prefixMatches(a.Rule.Prefix, path) { + continue + } + if ev != nil && !filterAllows(a.Rule, ev) { + continue + } + out = append(out, k) + } + return out +} + +func (s *Snapshot) filterMatching(keys []s3lifecycle.ActionKey, ev *Event) []s3lifecycle.ActionKey { + out := make([]s3lifecycle.ActionKey, 0, len(keys)) + for _, k := range keys { + a := s.actions[k] + if a == nil || !a.IsActive() || a.Bucket != ev.Bucket { + continue + } + if !prefixMatches(a.Rule.Prefix, ev.Path) { + continue + } + if !filterAllows(a.Rule, ev) { + continue + } + switch k.ActionKind { + case s3lifecycle.ActionKindAbortMPU: + if !ev.IsMPUInit { + continue + } + case s3lifecycle.ActionKindExpiredDeleteMarker: + if !ev.IsDeleteMarker || !ev.IsLatest { + continue + } + } + out = append(out, k) + } + return out +} + +func prefixMatches(rulePrefix, path string) bool { + if rulePrefix == "" { + return true + } + return strings.HasPrefix(path, rulePrefix) +} + +func filterAllows(rule *s3lifecycle.Rule, ev *Event) bool { + if rule.FilterSizeGreaterThan > 0 && ev.Size <= rule.FilterSizeGreaterThan { + return false + } + if rule.FilterSizeLessThan > 0 && ev.Size >= rule.FilterSizeLessThan { + return false + } + for k, v := range rule.FilterTags { + if got, ok := ev.Tags[k]; !ok || got != v { + return false + } + } + return true +} diff --git a/weed/s3api/s3lifecycle/engine/match_test.go b/weed/s3api/s3lifecycle/engine/match_test.go new file mode 100644 index 000000000..6009db9a6 --- /dev/null +++ b/weed/s3api/s3lifecycle/engine/match_test.go @@ -0,0 +1,195 @@ +package engine + +import ( + "reflect" + "sort" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// activeAll returns a PriorStates map that activates every action of the +// given canonical rules under bucket. +func activeAll(bucket string, rules []*s3lifecycle.Rule) map[s3lifecycle.ActionKey]PriorState { + out := map[s3lifecycle.ActionKey]PriorState{} + for _, r := range rules { + rh := s3lifecycle.RuleHash(r) + for _, k := range s3lifecycle.RuleActionKinds(r) { + out[s3lifecycle.ActionKey{Bucket: bucket, RuleHash: rh, ActionKind: k}] = PriorState{BootstrapComplete: true} + } + } + return out +} + +func sortedKeys(keys []s3lifecycle.ActionKey) []s3lifecycle.ActionKey { + out := append([]s3lifecycle.ActionKey(nil), keys...) + sort.Slice(out, func(i, j int) bool { + return out[i].ActionKind < out[j].ActionKind + }) + return out +} + +func TestMatchOriginalWrite_DelayGroupRoutes(t *testing.T) { + r30 := ruleExpDays("a", "x/", 30) + r60 := ruleExpDays("b", "x/", 60) + rules := []*s3lifecycle.Rule{r30, r60} + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: rules}}, CompileOptions{PriorStates: activeAll("bk", rules)}) + + ev := &Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: "x/o"} + got30 := snap.MatchOriginalWrite(ev, 30*24*time.Hour) + if len(got30) != 1 || got30[0].ActionKind != s3lifecycle.ActionKindExpirationDays { + t.Fatalf("30d sweep should match r30, got %v", got30) + } + got60 := snap.MatchOriginalWrite(ev, 60*24*time.Hour) + if len(got60) != 1 { + t.Fatalf("60d sweep should match r60, got %v", got60) + } +} + +func TestMatchOriginalWrite_PrefixFilter(t *testing.T) { + r := ruleExpDays("a", "logs/", 30) + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{r}}}, CompileOptions{PriorStates: activeAll("bk", []*s3lifecycle.Rule{r})}) + + if got := snap.MatchOriginalWrite(&Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: "data/x"}, 30*24*time.Hour); len(got) != 0 { + t.Fatalf("non-matching prefix should reject, got %v", got) + } + if got := snap.MatchOriginalWrite(&Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: "logs/x"}, 30*24*time.Hour); len(got) != 1 { + t.Fatalf("matching prefix should fire, got %v", got) + } +} + +func TestMatchOriginalWrite_MarkActiveBecomesRoutable(t *testing.T) { + // pending_bootstrap actions are indexed but inactive: MatchOriginalWrite + // returns nothing. After MarkActive flips engineState, the same key is + // routable in subsequent matches without a recompile. + r := ruleExpDays("a", "x/", 30) + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{r}}}, CompileOptions{}) + ev := &Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: "x/o"} + if got := snap.MatchOriginalWrite(ev, 30*24*time.Hour); len(got) != 0 { + t.Fatalf("inactive action should not match, got %v", got) + } + snap.MarkActive(s3lifecycle.ActionKey{Bucket: "bk", RuleHash: s3lifecycle.RuleHash(r), ActionKind: s3lifecycle.ActionKindExpirationDays}) + if got := snap.MatchOriginalWrite(ev, 30*24*time.Hour); len(got) != 1 { + t.Fatalf("post-markActive should be routable, got %v", got) + } +} + +func TestMatchOriginalWrite_AbortMPUOnlyOnMPUInit(t *testing.T) { + r := &s3lifecycle.Rule{ + ID: "mpu", + Status: s3lifecycle.StatusEnabled, + Prefix: ".uploads/", + AbortMPUDaysAfterInitiation: 7, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{r}}}, CompileOptions{PriorStates: activeAll("bk", []*s3lifecycle.Rule{r})}) + + // Non-MPU event under .uploads/ prefix: filtered out by shape gating. + got := snap.MatchOriginalWrite(&Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: ".uploads/u1/", IsMPUInit: false}, 7*24*time.Hour) + if len(got) != 0 { + t.Fatalf("non-MPU event should be filtered, got %v", got) + } + got = snap.MatchOriginalWrite(&Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: ".uploads/u1/", IsMPUInit: true}, 7*24*time.Hour) + if len(got) != 1 { + t.Fatalf("MPU init should fire, got %v", got) + } +} + +func TestMatchPredicateChange_OnlyTagSensitiveActions(t *testing.T) { + rTag := &s3lifecycle.Rule{ + ID: "tag", + Status: s3lifecycle.StatusEnabled, + Prefix: "x/", + ExpirationDays: 30, + FilterTags: map[string]string{"env": "prod"}, + } + rPlain := ruleExpDays("plain", "x/", 30) + rules := []*s3lifecycle.Rule{rTag, rPlain} + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: rules}}, CompileOptions{PriorStates: activeAll("bk", rules)}) + + // Predicate-change event: should match only the predicate-sensitive rule. + ev := &Event{Shape: EventShapePredicateChange, Bucket: "bk", Path: "x/o", Tags: map[string]string{"env": "prod"}} + got := snap.MatchPredicateChange(ev) + if len(got) != 1 { + t.Fatalf("only the tag-sensitive rule should match, got %v", got) + } + rh := s3lifecycle.RuleHash(rTag) + if got[0].RuleHash != rh { + t.Fatalf("expected match on tag-rule") + } + + // Wrong shape: never matches. + if got := snap.MatchPredicateChange(&Event{Shape: EventShapeOriginalWrite, Bucket: "bk", Path: "x/o"}); len(got) != 0 { + t.Fatalf("wrong shape should be empty, got %v", got) + } +} + +func TestMatchPath_BootstrapWalkerSeesAllActiveActions(t *testing.T) { + // One rule with three actions (90d Expiration, 30d NoncurrentDays, + // 7d AbortMPU). MatchPath returns every active action whose filter + // matches the path; the bootstrap walker iterates these and calls + // EvaluateAction per kind for the entry. + rule := &s3lifecycle.Rule{ + ID: "multi", + Status: s3lifecycle.StatusEnabled, + Prefix: "x/", + ExpirationDays: 90, + NoncurrentVersionExpirationDays: 30, + AbortMPUDaysAfterInitiation: 7, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: activeAll("bk", []*s3lifecycle.Rule{rule})}) + + got := snap.MatchPath("bk", "x/obj", nil) + if len(got) != 3 { + t.Fatalf("want all 3 active actions, got %v", got) + } + wantKinds := []s3lifecycle.ActionKind{ + s3lifecycle.ActionKindExpirationDays, + s3lifecycle.ActionKindNoncurrentDays, + s3lifecycle.ActionKindAbortMPU, + } + gotKinds := []s3lifecycle.ActionKind{} + for _, k := range sortedKeys(got) { + gotKinds = append(gotKinds, k.ActionKind) + } + sort.Slice(wantKinds, func(i, j int) bool { return wantKinds[i] < wantKinds[j] }) + if !reflect.DeepEqual(gotKinds, wantKinds) { + t.Fatalf("want kinds %v, got %v", wantKinds, gotKinds) + } +} + +func TestMatchPath_PrefixMismatchExcludes(t *testing.T) { + rule := ruleExpDays("a", "logs/", 30) + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: activeAll("bk", []*s3lifecycle.Rule{rule})}) + if got := snap.MatchPath("bk", "data/x", nil); len(got) != 0 { + t.Fatalf("prefix mismatch should reject, got %v", got) + } +} + +func TestMatchPath_FilterRejectsByTag(t *testing.T) { + rule := &s3lifecycle.Rule{ + ID: "tag", + Status: s3lifecycle.StatusEnabled, + Prefix: "x/", + ExpirationDays: 30, + FilterTags: map[string]string{"env": "prod"}, + } + e := New() + snap := e.Compile([]CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{rule}}}, CompileOptions{PriorStates: activeAll("bk", []*s3lifecycle.Rule{rule})}) + // With ev passed, tag mismatch rejects. + ev := &Event{Bucket: "bk", Path: "x/obj", Tags: map[string]string{"env": "dev"}} + if got := snap.MatchPath("bk", "x/obj", ev); len(got) != 0 { + t.Fatalf("tag mismatch should reject, got %v", got) + } + // With ev nil (caller will fetch live state): prefix-only, returns the action. + if got := snap.MatchPath("bk", "x/obj", nil); len(got) != 1 { + t.Fatalf("ev=nil prefix-only path should return action, got %v", got) + } +} diff --git a/weed/s3api/s3lifecycle/engine/mode.go b/weed/s3api/s3lifecycle/engine/mode.go new file mode 100644 index 000000000..faffeba0f --- /dev/null +++ b/weed/s3api/s3lifecycle/engine/mode.go @@ -0,0 +1,57 @@ +package engine + +import ( + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// RuleMode mirrors the durable s3_lifecycle_pb.LifecycleState.RuleMode enum; +// the worker maps between them when it reads/writes durable state. +type RuleMode int + +const ( + ModeUnspecified RuleMode = iota + ModeEventDriven + ModeScanAtDate + ModeScanOnly + ModeDisabled + ModePendingBootstrap +) + +func (m RuleMode) String() string { + switch m { + case ModeEventDriven: + return "event_driven" + case ModeScanAtDate: + return "scan_at_date" + case ModeScanOnly: + return "scan_only" + case ModeDisabled: + return "disabled" + case ModePendingBootstrap: + return "pending_bootstrap" + default: + return "unspecified" + } +} + +// decideMode: disabled rule -> DISABLED; EXPIRATION_DATE -> SCAN_AT_DATE; +// reader-driven kind whose horizon exceeds retention -> SCAN_ONLY; else +// EVENT_DRIVEN. metaLogRetention=0 means unbounded (default), gate doesn't +// trip. +func decideMode(rule *s3lifecycle.Rule, kind s3lifecycle.ActionKind, metaLogRetention, bootstrapLookbackMin time.Duration) RuleMode { + if rule == nil || rule.Status != s3lifecycle.StatusEnabled { + return ModeDisabled + } + if kind == s3lifecycle.ActionKindExpirationDate { + return ModeScanAtDate + } + if metaLogRetention > 0 { + horizon := s3lifecycle.EventLogHorizon(rule, kind) + if horizon > 0 && metaLogRetention < horizon+bootstrapLookbackMin { + return ModeScanOnly + } + } + return ModeEventDriven +} diff --git a/weed/s3api/s3lifecycle/evaluate.go b/weed/s3api/s3lifecycle/evaluate.go index e4fa49cae..6de9cf750 100644 --- a/weed/s3api/s3lifecycle/evaluate.go +++ b/weed/s3api/s3lifecycle/evaluate.go @@ -5,30 +5,10 @@ import ( "time" ) -// EvaluateAction decides whether the (rule, kind) compiled action fires for -// info at the given wall-clock time. Returns ActionNone when the rule is -// disabled, the filter rejects, the object shape doesn't match this kind, or -// the kind isn't yet due. -// -// One XML rule may declare multiple actions in parallel. The engine compiles -// each into its own ActionKey and calls EvaluateAction once per (rule, kind, -// entry); sibling actions are independent and may produce different verdicts -// for the same entry. Callers that need to evaluate every action of a rule -// iterate `RuleActionKinds(rule)` and call this helper per kind. -// -// Action selection by object shape per kind: -// -// IsMPUInit + ActionKindAbortMPU -> AbortIncompleteMultipartUpload -// IsLatest && IsDeleteMarker + ActionKindExpiredDeleteMarker -> ExpiredObjectDeleteMarker (sole survivor) -// IsLatest + ActionKindExpirationDays -> DeleteObject (Days threshold) -// IsLatest + ActionKindExpirationDate -> DeleteObject (date threshold) -// non-current + ActionKindNoncurrentDays -> DeleteVersion (Days + NewerNoncurrent retention) -// non-current + ActionKindNewerNoncurrent -> DeleteVersion (count-only) -// -// Per AWS S3 semantics, a non-current delete marker is treated as a regular -// non-current version under NONCURRENT_DAYS / NEWER_NONCURRENT. Only the -// *current* delete marker (and only as sole survivor) routes to -// EXPIRED_DELETE_MARKER. +// EvaluateAction returns whether the (rule, kind) action fires for info at +// now. A non-current delete marker is just another non-current version under +// NONCURRENT_DAYS / NEWER_NONCURRENT; only a current sole-survivor marker +// routes to EXPIRED_DELETE_MARKER. func EvaluateAction(rule *Rule, kind ActionKind, info *ObjectInfo, now time.Time) EvalResult { none := EvalResult{Action: ActionNone} if rule == nil || info == nil || rule.Status != StatusEnabled { @@ -89,18 +69,21 @@ func EvaluateAction(rule *Rule, kind ActionKind, info *ObjectInfo, now time.Time if now.Before(due) { return none } - if rule.NewerNoncurrentVersions > 0 && info.NoncurrentIndex < rule.NewerNoncurrentVersions { - return none + // nil index = can't evaluate retention; safety-scan revisits. + if rule.NewerNoncurrentVersions > 0 { + if info.NoncurrentIndex == nil || *info.NoncurrentIndex < rule.NewerNoncurrentVersions { + return none + } } return EvalResult{Action: ActionDeleteVersion, RuleID: rule.ID} case ActionKindNewerNoncurrent: - // Pure count-based: only when NoncurrentDays is unset (when paired, - // the rule expands to NONCURRENT_DAYS instead — see RuleActionKinds). + // Count-only; when paired with NoncurrentDays the rule expands to + // NONCURRENT_DAYS instead (RuleActionKinds). if info.IsLatest || rule.NoncurrentVersionExpirationDays > 0 || rule.NewerNoncurrentVersions <= 0 { return none } - if info.NoncurrentIndex < rule.NewerNoncurrentVersions { + if info.NoncurrentIndex == nil || *info.NoncurrentIndex < rule.NewerNoncurrentVersions { return none } return EvalResult{Action: ActionDeleteVersion, RuleID: rule.ID} diff --git a/weed/s3api/s3lifecycle/evaluate_test.go b/weed/s3api/s3lifecycle/evaluate_test.go index b36866ba7..ea59b9dd6 100644 --- a/weed/s3api/s3lifecycle/evaluate_test.go +++ b/weed/s3api/s3lifecycle/evaluate_test.go @@ -5,6 +5,11 @@ import ( "time" ) +// idx is a small helper for tests that need to express NoncurrentIndex as +// *int (production callers either compute it or leave it nil for current +// versions; tests want a one-liner). +func idx(i int) *int { return &i } + func mustTime(t *testing.T, s string) time.Time { t.Helper() tm, err := time.Parse(time.RFC3339, s) @@ -142,7 +147,7 @@ func TestEvaluateAction_NoncurrentVersionDays(t *testing.T) { IsLatest: false, ModTime: mustTime(t, "2023-01-01T00:00:00Z"), SuccessorModTime: successor, - NoncurrentIndex: 0, + NoncurrentIndex: idx(0), } if got := EvaluateAction(rule, ActionKindNoncurrentDays, info, successor.AddDate(0, 0, 29)); got.Action != ActionNone { t.Fatalf("not due, got %v", got) @@ -155,24 +160,56 @@ func TestEvaluateAction_NoncurrentVersionDays(t *testing.T) { func TestEvaluateAction_NoncurrentDaysFallsBackToModTime(t *testing.T) { rule := &Rule{Status: StatusEnabled, NoncurrentVersionExpirationDays: 30} mod := mustTime(t, "2024-01-01T00:00:00Z") - info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: mod, NoncurrentIndex: 0} + info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: mod, NoncurrentIndex: idx(0)} if got := EvaluateAction(rule, ActionKindNoncurrentDays, info, mod.AddDate(0, 0, 30)); got.Action != ActionDeleteVersion { t.Fatalf("expected fallback to ModTime, got %v", got) } } +func TestEvaluateAction_NoncurrentDays_WithKeepN_NilIndexIsNoOp(t *testing.T) { + // Pointer-migration safety: a nil NoncurrentIndex paired with a + // keep-N retention threshold must short-circuit to ActionNone rather + // than guess at the version's position in the keep window. + rule := &Rule{Status: StatusEnabled, NoncurrentVersionExpirationDays: 30, NewerNoncurrentVersions: 2} + successor := mustTime(t, "2024-01-01T00:00:00Z") + info := &ObjectInfo{ + Key: "a", + IsLatest: false, + ModTime: successor, + SuccessorModTime: successor, + NoncurrentIndex: nil, + } + now := successor.AddDate(0, 0, 30) + if got := EvaluateAction(rule, ActionKindNoncurrentDays, info, now); got.Action != ActionNone { + t.Fatalf("nil index with keep-N must be no-op, got %v", got) + } +} + +func TestEvaluateAction_NewerNoncurrent_NilIndexIsNoOp(t *testing.T) { + rule := &Rule{Status: StatusEnabled, NewerNoncurrentVersions: 3} + info := &ObjectInfo{ + Key: "a", + IsLatest: false, + ModTime: mustTime(t, "2024-01-01T00:00:00Z"), + NoncurrentIndex: nil, + } + if got := EvaluateAction(rule, ActionKindNewerNoncurrent, info, mustTime(t, "2025-01-01T00:00:00Z")); got.Action != ActionNone { + t.Fatalf("nil index pure-count must be no-op, got %v", got) + } +} + func TestEvaluateAction_NewerNoncurrentCountOnly(t *testing.T) { rule := &Rule{Status: StatusEnabled, NewerNoncurrentVersions: 3} now := mustTime(t, "2025-01-01T00:00:00Z") mod := mustTime(t, "2024-01-01T00:00:00Z") for i := 0; i < 3; i++ { - info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: mod, NoncurrentIndex: i} + info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: mod, NoncurrentIndex: idx(i)} if got := EvaluateAction(rule, ActionKindNewerNoncurrent, info, now); got.Action != ActionNone { t.Fatalf("idx=%d should be retained, got %v", i, got) } } - info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: mod, NoncurrentIndex: 3} + info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: mod, NoncurrentIndex: idx(3)} if got := EvaluateAction(rule, ActionKindNewerNoncurrent, info, now); got.Action != ActionDeleteVersion { t.Fatalf("idx=3 should fire, got %v", got) } @@ -184,12 +221,12 @@ func TestEvaluateAction_NoncurrentDaysAndCount(t *testing.T) { now := successor.AddDate(0, 0, 30) for i := 0; i < 2; i++ { - info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: successor, SuccessorModTime: successor, NoncurrentIndex: i} + info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: successor, SuccessorModTime: successor, NoncurrentIndex: idx(i)} if got := EvaluateAction(rule, ActionKindNoncurrentDays, info, now); got.Action != ActionNone { t.Fatalf("idx=%d kept by NewerNoncurrent retention, got %v", i, got) } } - info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: successor, SuccessorModTime: successor, NoncurrentIndex: 2} + info := &ObjectInfo{Key: "a", IsLatest: false, ModTime: successor, SuccessorModTime: successor, NoncurrentIndex: idx(2)} if got := EvaluateAction(rule, ActionKindNoncurrentDays, info, now); got.Action != ActionDeleteVersion { t.Fatalf("idx=2 satisfies both, got %v", got) } diff --git a/weed/s3api/s3lifecycle/event_log_horizon.go b/weed/s3api/s3lifecycle/event_log_horizon.go index 2f1cda21b..7c909cc74 100644 --- a/weed/s3api/s3lifecycle/event_log_horizon.go +++ b/weed/s3api/s3lifecycle/event_log_horizon.go @@ -2,24 +2,11 @@ package s3lifecycle import "time" -// EventLogHorizon returns the maximum age of an event the reader needs to -// observe to drive the (rule, kind) compiled action. Used by the retention -// mode gate: if metaLogRetention < EventLogHorizon(rule, kind) + -// bootstrapLookbackMin, this action is promoted to scan_only with -// degraded_reason=RETENTION_BELOW_HORIZON. -// -// One XML rule may declare multiple actions with different horizons (a 90d -// EXPIRATION_DAYS sibling alongside a 7d ABORT_MPU); the gate runs per -// compiled action so each can degrade independently. -// -// Per-kind values: -// -// EXPIRATION_DAYS -> rule.ExpirationDays -// NONCURRENT_DAYS -> rule.NoncurrentVersionExpirationDays -// ABORT_MPU -> rule.AbortMPUDaysAfterInitiation -// NEWER_NONCURRENT -> SmallDelay (count-only retention; immediate at flip) -// EXPIRED_DELETE_MARKER -> SmallDelay (immediate when sole survivor) -// EXPIRATION_DATE -> 0 (date kind bypasses the gate) +// EventLogHorizon returns the max event age the reader needs to drive the +// (rule, kind) action. Used by the retention mode gate: when +// metaLogRetention < EventLogHorizon + bootstrapLookbackMin, the action is +// promoted to scan_only. EXPIRATION_DATE returns 0 (date kind bypasses); +// count / immediate kinds return SmallDelay. func EventLogHorizon(rule *Rule, kind ActionKind) time.Duration { if rule == nil { return 0 diff --git a/weed/s3api/s3lifecycle/min_trigger_age.go b/weed/s3api/s3lifecycle/min_trigger_age.go index fcac7d0a8..56779ae09 100644 --- a/weed/s3api/s3lifecycle/min_trigger_age.go +++ b/weed/s3api/s3lifecycle/min_trigger_age.go @@ -2,14 +2,9 @@ package s3lifecycle import "time" -// MinTriggerAge returns the day threshold defined by `kind` on `rule`. Used by -// the safety-scan cadence as `max(MinTriggerAge(rule, kind), kindFloor)` — -// see the per-kind cadence table in the design doc. Returns 0 when the kind -// has no day-style threshold (date / count / immediate kinds), in which case -// the caller's kind-floor is the cadence directly. -// -// One XML rule may declare multiple actions; this helper takes a kind so the -// cadence is computed independently per compiled action. +// MinTriggerAge returns the day threshold defined by kind on rule, or 0 if +// the kind has no day-style threshold (date / count / immediate). Callers +// use it as max(MinTriggerAge, kindFloor) when computing safety-scan cadence. func MinTriggerAge(rule *Rule, kind ActionKind) time.Duration { if rule == nil { return 0 diff --git a/weed/s3api/s3lifecycle/rule.go b/weed/s3api/s3lifecycle/rule.go index cbf2a0135..5903c83f8 100644 --- a/weed/s3api/s3lifecycle/rule.go +++ b/weed/s3api/s3lifecycle/rule.go @@ -2,112 +2,73 @@ package s3lifecycle import "time" -// Rule is a flattened, evaluator-friendly representation of an S3 lifecycle rule. -// Callers convert from the XML-parsed s3api.Rule (which has nested structs with -// set-flags for conditional XML marshaling) to this type. +// Rule is the flat representation built from the XML-parsed s3api.Rule via +// s3api.LifecycleToCanonical. type Rule struct { ID string - Status string // "Enabled" or "Disabled" + Status string // "Enabled" | "Disabled" - // Prefix filter (from Rule.Prefix or Rule.Filter.Prefix or Rule.Filter.And.Prefix). Prefix string - // Expiration for current versions. ExpirationDays int ExpirationDate time.Time ExpiredObjectDeleteMarker bool - // Expiration for non-current versions. NoncurrentVersionExpirationDays int NewerNoncurrentVersions int - // Abort incomplete multipart uploads. AbortMPUDaysAfterInitiation int - // Tag filter (from Rule.Filter.Tag or Rule.Filter.And.Tags). FilterTags map[string]string - // Size filters. + // Zero is "not set"; can't represent an explicit + // 0 exclusion of empty objects. FilterSizeGreaterThan int64 FilterSizeLessThan int64 } -// ObjectInfo is the metadata about an object that the evaluator uses to -// determine which lifecycle action applies. Callers build this from filer -// entry attributes and extended metadata. type ObjectInfo struct { - // Key is the object key relative to the bucket root. - Key string - - // ModTime is the object's modification time (entry.Attributes.Mtime). - ModTime time.Time - - // Size is the object size in bytes (entry.Attributes.FileSize). - Size int64 - - // IsLatest is true if this is the current version of the object. - IsLatest bool - - // IsDeleteMarker is true if this entry is an S3 delete marker. + Key string + ModTime time.Time + Size int64 + IsLatest bool IsDeleteMarker bool + NumVersions int - // NumVersions is the total number of versions for this object key, - // including delete markers. Used for ExpiredObjectDeleteMarker evaluation. - NumVersions int - - // SuccessorModTime is the creation time of the version that replaced - // this one (making it non-current). Derived from the successor's version - // ID timestamp. Zero value for the latest version. SuccessorModTime time.Time - // NoncurrentIndex is the 0-based position among non-current versions - // sorted newest-first (0 = newest non-current version). Used by - // NewerNoncurrentVersions evaluation. -1 or unset for current versions. - NoncurrentIndex int + // NoncurrentIndex: 0-based among non-current versions, newest first. + // nil = current or not yet computed; count-based retention returns + // ActionNone rather than guess. Pointer so valid 0 doesn't collide + // with zero-value "uninitialised". + NoncurrentIndex *int - // Tags are the object's user-defined tags, extracted from the entry's - // Extended metadata (keys prefixed with "X-Amz-Tagging-"). Tags map[string]string - // IsMPUInit is true when this entry represents an in-flight multipart - // upload init under /.uploads//. The evaluator routes - // these entries to the AbortIncompleteMultipartUpload action shape. - // ModTime carries the upload's initiation time when this is set. + // IsMPUInit: in-flight upload under .uploads//; ModTime is init time. IsMPUInit bool } -// Status values for Rule.Status. const ( StatusEnabled = "Enabled" StatusDisabled = "Disabled" ) -// SmallDelay is the lookback used for predicate-change events and as the -// event-log horizon for count-based / immediate rule kinds (NewerNoncurrent, -// ExpiredObjectDeleteMarker). A small fixed value avoids races with in-flight -// writes without forcing the reader to keep deep history. +// SmallDelay is the lookback for predicate-change events and the event-log +// horizon for count / immediate kinds. const SmallDelay = time.Minute -// Action represents the lifecycle action to take on an object. type Action int const ( - // ActionNone means no lifecycle rule applies. ActionNone Action = iota - // ActionDeleteObject deletes the current version of the object. ActionDeleteObject - // ActionDeleteVersion deletes a specific non-current version. ActionDeleteVersion - // ActionExpireDeleteMarker removes a delete marker that is the sole remaining version. ActionExpireDeleteMarker - // ActionAbortMultipartUpload aborts an incomplete multipart upload. ActionAbortMultipartUpload ) -// EvalResult is the output of lifecycle rule evaluation. type EvalResult struct { - // Action is the lifecycle action to take. Action Action - // RuleID is the ID of the rule that triggered this action. RuleID string } diff --git a/weed/s3api/s3lifecycle/rule_hash.go b/weed/s3api/s3lifecycle/rule_hash.go index 042976b51..fee893558 100644 --- a/weed/s3api/s3lifecycle/rule_hash.go +++ b/weed/s3api/s3lifecycle/rule_hash.go @@ -7,33 +7,18 @@ import ( "time" ) -// RuleHash returns the first 8 bytes of sha256 over a canonicalized rule -// representation. The hash is stable across: -// -// - tag-key reorder (FilterTags is sorted before hashing) -// - rule.ID changes (ID is excluded — it's display-only) -// - rule.Status flips (Enabled <-> Disabled — state continuity preserved -// across operator toggles) -// -// Prefix is hashed verbatim: "logs" and "logs/" produce different hashes -// because they match different objects under literal prefix semantics -// (strings.HasPrefix). Collapsing them would let an edit silently reuse -// per-rule durable state for a rule that no longer matches the same set. -// -// Different action shapes (different days, different filter, different -// action types) hash to different values. -// -// Encoding is length-prefixed to avoid delimiter ambiguity: a tag value -// containing "=" or "\n" or a prefix containing the field-tag separator -// must not be able to forge a different tuple that hashes the same. Each -// scalar is written as ` `. +// RuleHash returns the first 8 bytes of sha256 over a length-prefixed +// canonical encoding. Stable across tag-key reorder, ID renames, and Status +// flips. Prefix is hashed verbatim — "logs" and "logs/" match different +// objects, so they must hash differently. Length prefixing prevents a +// forged value (e.g. a tag containing "=") from colliding with a legitimate +// tuple. func RuleHash(rule *Rule) [8]byte { if rule == nil { var zero [8]byte return zero } h := sha256.New() - // Filter. writeBytes(h, fieldPrefix, []byte(rule.Prefix)) tagKeys := make([]string, 0, len(rule.FilterTags)) for k := range rule.FilterTags { @@ -47,7 +32,6 @@ func RuleHash(rule *Rule) [8]byte { } writeInt64(h, fieldSizeGT, rule.FilterSizeGreaterThan) writeInt64(h, fieldSizeLT, rule.FilterSizeLessThan) - // Actions. writeInt64(h, fieldExpDays, int64(rule.ExpirationDays)) writeBytes(h, fieldExpDate, []byte(canonicalTime(rule.ExpirationDate))) writeBool(h, fieldExpDeleteMarker, rule.ExpiredObjectDeleteMarker) @@ -61,8 +45,8 @@ func RuleHash(rule *Rule) [8]byte { return out } -// Field tags namespace each scalar so an attacker can't substitute one -// string for another and re-collide. Values are arbitrary but stable. +// Field tags namespace each scalar so a forged string can't substitute for +// another in a way that hashes the same. const ( fieldPrefix byte = 0x01 fieldTagCount byte = 0x02