From 7f2b20d577f5e2b10d53b23b0d81b4639c875fe3 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 7 May 2026 15:00:49 -0700 Subject: [PATCH] =?UTF-8?q?feat(s3/lifecycle):=20policy=20engine=20?= =?UTF-8?q?=E2=80=94=20XML=20conversion,=20Compile,=20decideMode,=20Match?= =?UTF-8?q?=20(#9348)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): XML lifecycle config to canonical Rule LifecycleToCanonical takes a parsed *Lifecycle and returns []*s3lifecycle.Rule, the flat shape the engine compiles against. Filter resolution mirrors AWS: sub-elements (Prefix + Tags + size filters) flatten into the canonical Rule's individual fields; single filter populates FilterTags with one entry; filter takes precedence over the rule's top-level . Multi-action rules (Expiration + NoncurrentVersion + AbortMPU on the same XML ) populate every action field they declare. RuleActionKinds expands the canonical rule into its compiled actions downstream. * feat(s3/lifecycle): engine snapshot skeleton + ActionKey type Defines s3lifecycle.ActionKey{rule_hash, action_kind} as the engine's primary identity, and adds the engine package's Snapshot type. Snapshot is immutable after Compile (atomic-swapped on rebuild) and holds the ActionKey-keyed routing indexes: - originalDelayGroups: map[time.Duration][]ActionKey - predicateActions: []ActionKey - dateActions: map[ActionKey]time.Time - actions: map[ActionKey]*CompiledAction CompiledAction.engineState is an atomic.Uint32 so MarkActive (called after the durable bootstrap_complete + mode write commits) is visible to in-flight reader passes without a recompile. The reader filters on IsActive() before dispatching, so stale-snapshot dispatches are prevented. No callers yet; downstream commits add Compile, decideMode, and the Match functions. * feat(s3/lifecycle): decideMode + retention gate decideMode picks the scheduling mode for one (rule, kind) compiled action. Disabled rule -> DISABLED; EXPIRATION_DATE -> SCAN_AT_DATE; reader-driven kind whose eventLogHorizon + bootstrapLookbackMin exceeds metaLogRetention -> SCAN_ONLY; otherwise EVENT_DRIVEN. The gate runs per (rule, kind), so a 90d ExpirationDays sibling can degrade to scan_only while its 7d AbortMPU sibling stays active. MetaLogRetention=0 is treated as "unbounded" — matches the SeaweedFS default (Phase 0 verified that meta-log files are written without TtlSec by default), so the gate doesn't trip until an operator opts in to volume-TTL pruning of /topics/.system/log/. RuleMode is a Go-level enum here, separate from the wire-form LifecycleState.RuleMode in the proto package; the worker maps between them when reading/writing the durable state file. * feat(s3/lifecycle): Compile builds the engine snapshot per-action Compile produces a fresh Snapshot from per-bucket canonical rules. Each input rule expands into N CompiledActions via RuleActionKinds; mode comes from decideMode; activation requires both bootstrap_complete (from PriorStates) and mode==EVENT_DRIVEN. Routing indexes are populated by mode: - SCAN_AT_DATE: always indexed in dateActions (detector schedules at rule.date regardless of bootstrap status; the action runs once on the date and is then done). - EVENT_DRIVEN + active: indexed in originalDelayGroups (and in predicateActions when the rule has tag/size filters). - SCAN_ONLY / DISABLED / pending_bootstrap: not indexed; safety-scan tick or operator action handle these. snapshot_id is monotonic per process; pending writes stamp it. The new snapshot replaces the engine's atomic pointer; in-flight reader passes continue against their loaded snapshot. Tests cover: single-action rule, multi-action expansion (one rule -> three CompiledActions with three distinct delay groups), pending bootstrap exclusion from indexes, retention gate, sibling actions degrading independently under partial retention, ExpirationDate path, disabled rule, MarkActive flipping IsActive(), Compile producing monotonic snapshot ids. * feat(s3/lifecycle): MatchOriginalWrite / MatchPredicateChange / MatchPath The reader feeds events through the engine's match functions to find the active ActionKeys whose filter applies. The minimal Event shape the engine takes (bucket, path, tags, size, IsLatest, IsDeleteMarker, IsMPUInit) keeps engine free of filer_pb dependencies; the reader extracts these fields from the persisted *filer_pb.LogEntry payload in Phase 3. - MatchOriginalWrite: per-delay-group sweep entry. Filters on shape = EventShapeOriginalWrite, prefix, tag, size, then per-kind shape gating (ABORT_MPU only on IsMPUInit; EXPIRED_DELETE_MARKER only on IsLatest+IsDeleteMarker). - MatchPredicateChange: single near-now sweep. Returns only the predicate-sensitive subset of active ActionKeys. - MatchPath: bucket-level walker entry. Returns every active action whose filter matches; bootstrap iterates these per object and calls EvaluateAction per kind. All filter on a.IsActive() at routing time so MarkActive flips become visible without recompile. * fix(s3/lifecycle): scope ActionKey by bucket; defensive copies; tidy compile Three findings on the engine PR addressed: 1. Critical (cross-bucket collision): ActionKey was {RuleHash, ActionKind} only. Two buckets with rules whose XML is identical produce the same RuleHash; the second bucket's Compile would overwrite the first bucket's CompiledAction in snap.actions. Add Bucket to ActionKey so the engine's identity matches the on-disk path layout /etc/s3/lifecycle////. Regression test pins it. 2. Major (immutability leak): OriginalDelayGroups, PredicateActions, DateActions returned the snapshot's internal maps/slices by reference, letting an external caller mutate routing state and break the documented immutability contract. Return defensive copies. 3. Minor (redundant condition): mode==EVENT_DRIVEN already implies kind != EXPIRATION_DATE because decideMode routes the date kind to SCAN_AT_DATE. Drop the redundant check. Tests updated to construct ActionKey with the new Bucket field. * fix(s3/lifecycle): drop size filters from rulePredicateSensitive An object's size is immutable once written: any content change is a fresh write that flows through the original-write stream, not the predicate-change one. Tagging rules really can flip post-PUT (operator adds/removes a tag without rewriting), so they belong; size filters do not. Including size filters here was adding rules to predicateActions for no purpose — every predicate-change sweep would waste cycles re-evaluating size predicates that physically can't have changed. * perf(s3/lifecycle): pre-sort AllActions at Compile time Snapshot is immutable after Compile (engineState bit-flips don't change membership), so the (bucket, rule_hash, action_kind) ordering is stable for the snapshot's lifetime. Build the sorted slice once and serve every AllActions() call from it; drop the per-call sort.Slice. The bootstrap walker is the primary caller and may iterate this on every task entry. * docs(s3/lifecycle): note the FilterSizeGreaterThan=0 ambiguity Per AWS S3 spec, 0 explicitly excludes 0-byte objects, but with the int64 zero value as the unset sentinel we can't distinguish that from omitted-and-default. Document the limitation inline so a future deployment that needs the distinction can switch to *int64 (or a paired set-bool) and update the matchers / RuleHash accordingly. Not fixing now: the explicit-zero configuration is unusual, the canonical Rule shape mirrors the same zero-as-unset convention as s3api.Filter, and a structural fix touches every filter-using site (evaluator, due_at, match, RuleHash). * fix(s3/lifecycle): make ObjectInfo.NoncurrentIndex *int The previous int field had a zero-value collision: 0 is both "newest non-current version" (a valid index) and "uninitialised by ObjectInfo{} literal." A caller who built &ObjectInfo{IsLatest: false} without explicitly setting NoncurrentIndex would have it implicitly read as "newest non-current," and the count-based NewerNoncurrent retention would use that bogus 0 to decide eligibility. Switch to *int so nil is explicitly "not a non-current version / index not yet computed." The evaluator's NoncurrentDays and NewerNoncurrent paths conservatively return ActionNone when the index is nil — the safety scan will revisit once the index is supplied. This removes a class of latent footguns in test setup and in any future code path that constructs ObjectInfo without a versioning-aware builder. idx() helper added in tests to keep the call sites a one-liner. * refactor(s3/lifecycle): trim narration from engine + helpers Drop "what" comments where well-named identifiers already say it (IsActive, MarkActive, AllActions, etc.); collapse multi-paragraph "why" docs to one-liners where the design rationale is already in the design doc. Keep WHY comments only at non-obvious load-bearing spots: the routing-index activation predicate, the *int rationale on NoncurrentIndex, the field-tag namespace in RuleHash, the SmallDelay horizon rule. Files: action_kind.go, rule.go, rule_hash.go, evaluate.go, due_at.go, min_trigger_age.go, event_log_horizon.go, engine/engine.go, engine/compile.go, engine/match.go, engine/mode.go. No behavior change; tests untouched and pass. * fix(s3/lifecycle): durable PriorState.Mode wins over decideMode PriorState.Mode was declared but never read; Compile recomputed mode via decideMode and stored that on every CompiledAction. Effect: an action durably persisted as SCAN_ONLY (lag fallback or operator pause) or DISABLED would silently re-promote to EVENT_DRIVEN on the next engine rebuild as soon as decideMode's XML+retention predicate said so. Defeats the durability of mode state. Use prior.Mode when set; fall through to decideMode only for new actions (no prior at all) and for legacy entries persisted before Mode existed (zero value). Regression test pins both branches. * fix(s3/lifecycle): MarkActive routability — index every EVENT_DRIVEN key MarkActive's documented contract was "flip visible without a recompile," but the routing indexes (originalDelayGroups, predicateActions) were only populated when active && mode == EVENT_DRIVEN at compile time. So a key compiled with BootstrapComplete=false would never enter the indexes; a later MarkActive flipped engineState but MatchOriginalWrite / MatchPredicateChange iterated the indexes and never saw the key. Only MatchPath (which walks bi.actionKeys) and DateActions worked. Index every EVENT_DRIVEN key regardless of `active`. The runtime IsActive() filter inside filterMatching already gates dispatch, so inactive entries are matched-but-not-fired; flipping MarkActive makes them routable without recompile, matching the documented contract. Tests updated: TestCompile_BootstrapPendingIndexedButInactive asserts the indexed-but-inactive shape; TestMatchOriginalWrite_MarkActiveBecomesRoutable asserts a MarkActive flip routes the next match. * test(s3/lifecycle): pin nil NoncurrentIndex no-op behavior Two regression tests for the *int pointer migration: nil index combined with NewerNoncurrent (either paired with NoncurrentDays or standalone) must short-circuit to ActionNone rather than guess at the version's position in the keep-N window. * refactor(s3/lifecycle): trim follow-up narration on engine + helpers Comments accumulated since the last sweep — the durable-Mode rationale, the MarkActive routability note, the routing-index doc, the NoncurrentIndex pointer rationale, and the EvaluateAction docblock. Trimmed each to one or two terse lines; the underlying contracts live in the design doc. * docs(s3/lifecycle): note CompileInput one-per-bucket invariant --- weed/s3api/s3api_lifecycle_canonical.go | 101 ++++++ weed/s3api/s3api_lifecycle_canonical_test.go | 196 ++++++++++++ weed/s3api/s3lifecycle/action_kind.go | 47 ++- weed/s3api/s3lifecycle/due_at.go | 13 +- weed/s3api/s3lifecycle/engine/compile.go | 126 ++++++++ weed/s3api/s3lifecycle/engine/engine.go | 120 +++++++ weed/s3api/s3lifecycle/engine/engine_test.go | 310 +++++++++++++++++++ weed/s3api/s3lifecycle/engine/match.go | 134 ++++++++ weed/s3api/s3lifecycle/engine/match_test.go | 195 ++++++++++++ weed/s3api/s3lifecycle/engine/mode.go | 57 ++++ weed/s3api/s3lifecycle/evaluate.go | 41 +-- weed/s3api/s3lifecycle/evaluate_test.go | 49 ++- weed/s3api/s3lifecycle/event_log_horizon.go | 23 +- weed/s3api/s3lifecycle/min_trigger_age.go | 11 +- weed/s3api/s3lifecycle/rule.go | 75 ++--- weed/s3api/s3lifecycle/rule_hash.go | 32 +- 16 files changed, 1354 insertions(+), 176 deletions(-) create mode 100644 weed/s3api/s3api_lifecycle_canonical.go create mode 100644 weed/s3api/s3api_lifecycle_canonical_test.go create mode 100644 weed/s3api/s3lifecycle/engine/compile.go create mode 100644 weed/s3api/s3lifecycle/engine/engine.go create mode 100644 weed/s3api/s3lifecycle/engine/engine_test.go create mode 100644 weed/s3api/s3lifecycle/engine/match.go create mode 100644 weed/s3api/s3lifecycle/engine/match_test.go create mode 100644 weed/s3api/s3lifecycle/engine/mode.go 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