diff --git a/weed/s3api/s3lifecycle/DESIGN.md b/weed/s3api/s3lifecycle/DESIGN.md index 315efed796..7ff33f707e 100644 --- a/weed/s3api/s3lifecycle/DESIGN.md +++ b/weed/s3api/s3lifecycle/DESIGN.md @@ -61,7 +61,8 @@ runShard(ctx, cfg, snap, runNow, shardID, events): lastWalkedNs = runNow.UnixNano() startTsNs = !found ? runNow - maxTTL : persisted.TsNs // steady state honors cursor - lastOK, _, drainErr = drainShardEvents(ctx, cfg, runNow, shardID, snap, startTsNs, events) + replayView = snap.ReplayView(retentionWindow) // minus walk-bound rules + lastOK, _, drainErr = drainShardEvents(ctx, cfg, runNow, shardID, replayView, startTsNs, events) save cursor (TsNs = lastOK, rsh, promoted, lastWalkedNs) ``` @@ -81,6 +82,7 @@ Walker invocations — three call sites, distinct conditions. `cfg.Walker` takes // In engine: func (e *Engine) Snapshot() *Snapshot func (s *Snapshot) RulesForShard(shardID int, retentionWindow time.Duration) (replay, walk *Snapshot) +func (s *Snapshot) ReplayView(retentionWindow time.Duration) *Snapshot func RecoveryView(s *Snapshot) *Snapshot func ReplayContentHash(s *Snapshot) [32]byte func PromotedHash(s *Snapshot, retentionWindow time.Duration) [32]byte @@ -102,6 +104,7 @@ Shared-by-pointer with the base: `Rule` definitions, predicate maps, `RuleHash` | View | Clone settings | Why | |---|---|---| | `replay` | `active = true`, `Mode = ModeEventDriven` | `router.Route` requires `ModeEventDriven`. Forced regardless of `prior.Mode`. | +| `replayView` | `active` preserved, `Mode = ModeEventDriven` | The drain's routing set: `replay` partition minus promoted rules, but without force-activating pre-bootstrap actions. Walk-bound kinds are shared by pointer. | | `walk` | `active = true`, `Mode` preserved | Walker accepts any non-`ModeDisabled` Mode. | | `recovery` | `active = true`, `Mode` preserved | Walker iterates all action clones. | diff --git a/weed/s3api/s3lifecycle/dailyrun/replay_view_test.go b/weed/s3api/s3lifecycle/dailyrun/replay_view_test.go new file mode 100644 index 0000000000..8adb1da419 --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/replay_view_test.go @@ -0,0 +1,67 @@ +package dailyrun + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRunShard_WalkBoundRuleDoesNotPinCursor: a bucket with a 3-day and a +// 30-day expiration rule and an 8-day retention window. The 30-day rule is +// walk-bound, so its object's PUT must not hold the replay cursor back; the +// cursor advances past it and the due 3-day objects on both sides dispatch. +// The 3-day rule also carries a durable scan_only mode: the replay view +// must rehabilitate it to event_driven the way RulesForShard does. +func TestRunShard_WalkBoundRuleDoesNotPinCursor(t *testing.T) { + shortRule := &s3lifecycle.Rule{ID: "short", Status: s3lifecycle.StatusEnabled, Prefix: "short/", ExpirationDays: 3} + longRule := &s3lifecycle.Rule{ID: "long", Status: s3lifecycle.StatusEnabled, Prefix: "long/", ExpirationDays: 30} + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: "b1", Rules: []*s3lifecycle.Rule{shortRule, longRule}}, + }, engine.CompileOptions{PriorStates: map[s3lifecycle.ActionKey]engine.PriorState{ + {Bucket: "b1", RuleHash: s3lifecycle.RuleHash(shortRule), ActionKind: s3lifecycle.ActionKindExpirationDays}: { + BootstrapComplete: true, Mode: engine.ModeScanOnly}, + {Bucket: "b1", RuleHash: s3lifecycle.RuleHash(longRule), ActionKind: s3lifecycle.ActionKindExpirationDays}: { + BootstrapComplete: true}, + }}) + snap := e.Snapshot() + for _, a := range snap.AllActions() { + snap.MarkActive(a.Key) + } + runNow := time.Unix(1_700_000_000, 0).UTC() + retention := s3lifecycle.DaysToDuration(8) + startTsNs := runNow.Add(-retention).UnixNano() + p := newMemPersister() + require.NoError(t, p.Save(context.Background(), 0, Cursor{ + TsNs: startTsNs, + RuleSetHash: engine.ReplayContentHash(snap), + PromotedHash: engine.PromotedHash(snap, retention), + LastWalkedNs: runNow.UnixNano(), + })) + put := func(key string, age time.Duration) *reader.Event { + at := runNow.Add(-age) + return &reader.Event{TsNs: at.UnixNano(), Bucket: "b1", Key: key, NewEntry: &filer_pb.Entry{ + Name: key, + Attributes: &filer_pb.FuseAttributes{Mtime: at.Unix(), FileSize: 1}, + }} + } + events := make(chan *reader.Event, 3) + events <- put("short/a", s3lifecycle.DaysToDuration(7)) + events <- put("long/a", s3lifecycle.DaysToDuration(6)) + last := put("short/b", s3lifecycle.DaysToDuration(5)) + events <- last + close(events) + client := &recordingClient{} + cfg := Config{Persister: p, Client: client, Lister: stubSiblingLister{}, RetentionWindow: retention} + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, events)) + assert.ElementsMatch(t, []string{"short/a", "short/b"}, client.seenObjects()) + got, _, _ := p.Load(context.Background(), 0) + assert.Equal(t, last.TsNs, got.TsNs, "cursor must advance past the walk-bound object's PUT") +} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 1488db6d51..beeacfa6d4 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -469,6 +469,12 @@ func validate(cfg Config) error { if cfg.SubscriptionReceiveTimeout < 0 { return fmt.Errorf("daily_run: negative SubscriptionReceiveTimeout %v (0 = default %v)", cfg.SubscriptionReceiveTimeout, defaultSubscriptionReceiveTimeout) } + // A declared retention window promotes long-TTL rules to the walk + // view; without a walker they lose their only dispatch path while + // the drain cursor still advances past their events. + if cfg.RetentionWindow > 0 && cfg.Walker == nil { + return errors.New("daily_run: RetentionWindow requires Walker (promoted rules would be silently skipped)") + } for _, sh := range cfg.Shards { if sh < 0 || sh >= s3lifecycle.ShardCount { return fmt.Errorf("daily_run: shard %d out of [0, %d)", sh, s3lifecycle.ShardCount) @@ -635,7 +641,10 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim startTsNs = runNow.Add(-maxTTL).UnixNano() } - lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, snap, startTsNs, events) + // Walk-bound rules are never due inside the retention window; keep them + // out of the drain so their matches don't pin the cursor. + replayView := snap.ReplayView(retentionWindow) + lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, replayView, startTsNs, events) // Fresh ctx: a wall-clock cap on the pass (the shell driver's // -runtime) cancels the drain's, and saving with it would drop the // cursor and re-replay from the same floor next pass. diff --git a/weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go b/weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go index cbf4b15607..377f3444ac 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go @@ -219,6 +219,19 @@ func TestValidate_RejectsNegativeSubscriptionReceiveTimeout(t *testing.T) { require.NoError(t, validate(cfg)) } +// RetentionWindow without a Walker would drop promoted rules' only +// dispatch path while the drain cursor still advances past their events. +func TestValidate_RejectsRetentionWindowWithoutWalker(t *testing.T) { + cfg := validatableConfig() + cfg.RetentionWindow = 24 * time.Hour + require.ErrorContains(t, validate(cfg), "RetentionWindow") + cfg.Walker = func(context.Context, *engine.Snapshot, int) error { return nil } + require.NoError(t, validate(cfg)) + cfg.Walker = nil + cfg.RetentionWindow = 0 + require.NoError(t, validate(cfg)) +} + // slowSavePersister stretches teardown past the caller's deadline. type slowSavePersister struct { *memPersister diff --git a/weed/s3api/s3lifecycle/engine/compile.go b/weed/s3api/s3lifecycle/engine/compile.go index 71bda62949..411b607bb2 100644 --- a/weed/s3api/s3lifecycle/engine/compile.go +++ b/weed/s3api/s3lifecycle/engine/compile.go @@ -80,6 +80,7 @@ func (e *Engine) Compile(inputs []CompileInput, opts CompileOptions) *Snapshot { Delay: s3lifecycle.MinTriggerAge(rule, kind), PredicateSensitive: rulePredicateSensitive(rule), Mode: mode, + bootstrapComplete: prior.BootstrapComplete, } if active { ca.markActive() diff --git a/weed/s3api/s3lifecycle/engine/engine.go b/weed/s3api/s3lifecycle/engine/engine.go index d7ad7e7c58..9695e46387 100644 --- a/weed/s3api/s3lifecycle/engine/engine.go +++ b/weed/s3api/s3lifecycle/engine/engine.go @@ -54,6 +54,7 @@ type CompiledAction struct { Delay time.Duration PredicateSensitive bool Mode RuleMode + bootstrapComplete bool engineState atomic.Uint32 } diff --git a/weed/s3api/s3lifecycle/engine/hashes.go b/weed/s3api/s3lifecycle/engine/hashes.go index 42815a1e5f..169b62e566 100644 --- a/weed/s3api/s3lifecycle/engine/hashes.go +++ b/weed/s3api/s3lifecycle/engine/hashes.go @@ -152,11 +152,7 @@ func PromotedHash(s *Snapshot, retentionWindow time.Duration) [32]byte { if !isReplayKind(k.ActionKind) { continue } - ttl := effectiveTTL(a) - // Mirror RulesForShard's partition predicate exactly: a replay - // kind lands in walk when ttl is 0 (malformed) or ttl > - // retentionWindow. PromotedHash hashes that walk-bound subset. - if ttl > 0 && ttl <= retentionWindow { + if !isPromoted(a, retentionWindow) { continue } items = append(items, hashItem{key: k, action: a}) diff --git a/weed/s3api/s3lifecycle/engine/views.go b/weed/s3api/s3lifecycle/engine/views.go index 7770ca668e..1c0d92802e 100644 --- a/weed/s3api/s3lifecycle/engine/views.go +++ b/weed/s3api/s3lifecycle/engine/views.go @@ -59,6 +59,7 @@ func cloneAction(src *CompiledAction, mode RuleMode, active bool) *CompiledActio Delay: src.Delay, PredicateSensitive: src.PredicateSensitive, Mode: mode, + bootstrapComplete: src.bootstrapComplete, } if active { dst.markActive() @@ -145,6 +146,17 @@ func effectiveTTL(a *CompiledAction) time.Duration { return 0 } +// isPromoted reports whether a replay-kind action belongs to the walk +// partition for retentionWindow. retentionWindow == 0 means "no retention +// info supplied": treat the partition as walk-only — promoting every rule +// into replay without retention proof would defeat the scan_only +// protection. ttl == 0 is a malformed rule (kind says replay but no TTL): +// also promoted so the walker can decide. +func isPromoted(a *CompiledAction, retentionWindow time.Duration) bool { + ttl := effectiveTTL(a) + return ttl <= 0 || ttl > retentionWindow +} + // RulesForShard partitions the base snapshot's compiled actions into a // replay view and a walk view. // @@ -187,14 +199,7 @@ func (s *Snapshot) RulesForShard(shardID int, retentionWindow time.Duration) (re continue } if isReplayKind(key.ActionKind) { - ttl := effectiveTTL(a) - // retentionWindow == 0 means "no retention info supplied" — the - // caller should treat the partition as walk-only to stay safe, - // since promoting every rule into replay without retention - // proof would defeat the scan_only protection. ttl == 0 is a - // malformed rule (kind says replay but no TTL): also routes to - // walk so the walker can decide. - if ttl > 0 && ttl <= retentionWindow { + if !isPromoted(a, retentionWindow) { replayActions[key] = cloneAction(a, ModeEventDriven, true) } else { walkActions[key] = cloneAction(a, a.Mode, true) @@ -209,6 +214,45 @@ func (s *Snapshot) RulesForShard(shardID int, retentionWindow time.Duration) (re return newView(s, replayActions), newView(s, walkActions) } +// ReplayView returns the snapshot the meta-log drain routes events against: +// the base snapshot minus the replay-kind actions promoted to the walk +// partition for retentionWindow. Promoted matches are never due within the +// replay window, and the drain holds a shard's cursor at the first event +// with a not-yet-due match, so one promoted match would pin the cursor +// until that object comes due; the walk view serves them instead. +// +// Actions are shared with the base by pointer so a MarkActive flip stays +// observable mid-drain. The exception is a replay-kind action still +// carrying ModeScanOnly: it is cloned as ModeEventDriven and reactivated +// when its bootstrap rendezvous already completed, for the same reason +// RulesForShard rewrites the mode — router.Route gates on event_driven, +// and scan_only is a retention verdict, not a bootstrap state. Disabled +// actions stay out. +// +// Returns nil when every action is promoted or disabled. +func (s *Snapshot) ReplayView(retentionWindow time.Duration) *Snapshot { + if s == nil { + return nil + } + actions := map[s3lifecycle.ActionKey]*CompiledAction{} + for key, a := range s.actions { + if a == nil || a.Mode == ModeDisabled { + continue + } + if isReplayKind(key.ActionKind) { + if isPromoted(a, retentionWindow) { + continue + } + if a.Mode == ModeScanOnly { + actions[key] = cloneAction(a, ModeEventDriven, a.IsActive() || a.bootstrapComplete) + continue + } + } + actions[key] = a + } + return newView(s, actions) +} + // RecoveryView returns a *Snapshot that contains a clone of every action in // the base, with active=true regardless of compile-time IsActive/ // BootstrapComplete and Mode preserved as-is. The recovery walker uses this diff --git a/weed/s3api/s3lifecycle/engine/views_test.go b/weed/s3api/s3lifecycle/engine/views_test.go index 7c2a9973bc..cd1e61774f 100644 --- a/weed/s3api/s3lifecycle/engine/views_test.go +++ b/weed/s3api/s3lifecycle/engine/views_test.go @@ -235,6 +235,70 @@ func TestRulesForShard_ZeroRetentionRoutesAllReplayKindsToWalk(t *testing.T) { assert.Len(t, walk.actions, 1) } +func TestReplayView_PartitionAndScanOnlyRehabilitation(t *testing.T) { + short := &s3lifecycle.Rule{ID: "short", Status: s3lifecycle.StatusEnabled, ExpirationDays: 3} + long := &s3lifecycle.Rule{ID: "long", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30} + date := &s3lifecycle.Rule{ID: "date", Status: s3lifecycle.StatusEnabled, ExpirationDate: time.Now().Add(48 * time.Hour)} + off := &s3lifecycle.Rule{ID: "off", Status: s3lifecycle.StatusDisabled, ExpirationDays: 7} + pending := &s3lifecycle.Rule{ID: "pending", Status: s3lifecycle.StatusEnabled, ExpirationDays: 5} + locked := &s3lifecycle.Rule{ID: "locked", Status: s3lifecycle.StatusEnabled, ExpirationDays: 4} + shortKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(short), ActionKind: s3lifecycle.ActionKindExpirationDays} + longKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(long), ActionKind: s3lifecycle.ActionKindExpirationDays} + pendingKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(pending), ActionKind: s3lifecycle.ActionKindExpirationDays} + lockedKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(locked), ActionKind: s3lifecycle.ActionKindExpirationDays} + snap := New().Compile( + []CompileInput{{Bucket: "b1", Rules: []*s3lifecycle.Rule{short, long, date, off, pending, locked}}}, + CompileOptions{PriorStates: map[s3lifecycle.ActionKey]PriorState{ + shortKey: {BootstrapComplete: true, Mode: ModeScanOnly}, + longKey: {BootstrapComplete: true}, + lockedKey: {Mode: ModeScanOnly}, // demoted before bootstrap finished + }}, + ) + + rv := snap.ReplayView(s3lifecycle.DaysToDuration(8)) + require.NotNil(t, rv) + + // The 30d rule is promoted to walk; the drain must not route it. + assert.Nil(t, rv.actions[longKey]) + // Disabled rules stay out entirely; short+date+pending+locked remain. + assert.Len(t, rv.actions, 4) + // ScanOnly lock-in rehabilitates to event_driven once bootstrap + // completed; the base is untouched. + clone := rv.actions[shortKey] + require.NotNil(t, clone) + assert.NotSame(t, snap.actions[shortKey], clone) + assert.Equal(t, ModeEventDriven, clone.Mode) + assert.True(t, clone.IsActive()) + assert.Equal(t, ModeScanOnly, snap.actions[shortKey].Mode) + // ScanOnly without a completed bootstrap stays inactive — the drain + // must not dispatch rules whose rendezvous never ran. + lockedClone := rv.actions[lockedKey] + require.NotNil(t, lockedClone) + assert.Equal(t, ModeEventDriven, lockedClone.Mode) + assert.False(t, lockedClone.IsActive()) + // Pre-bootstrap event-driven actions keep active=false and share the + // base pointer so a mid-pass MarkActive stays observable. + assert.Same(t, snap.actions[pendingKey], rv.actions[pendingKey]) + assert.False(t, rv.actions[pendingKey].IsActive()) + // Walk-bound kinds share the base pointer, Mode preserved. + dateKey := s3lifecycle.ActionKey{Bucket: "b1", RuleHash: s3lifecycle.RuleHash(date), ActionKind: s3lifecycle.ActionKindExpirationDate} + assert.Same(t, snap.actions[dateKey], rv.actions[dateKey]) + assert.Equal(t, ModeScanAtDate, rv.actions[dateKey].Mode) +} + +func TestReplayView_AllPromotedReturnsNil(t *testing.T) { + rule := &s3lifecycle.Rule{ID: "long", Status: s3lifecycle.StatusEnabled, ExpirationDays: 90} + snap := buildSnapshotForViews(t, "b1", rule) + assert.Nil(t, snap.ReplayView(s3lifecycle.DaysToDuration(8))) + assert.Nil(t, snap.ReplayView(0)) + assert.Nil(t, snap.ReplayView(0)) +} + +func TestReplayView_NilSnapshot(t *testing.T) { + var s *Snapshot + assert.Nil(t, s.ReplayView(0)) +} + func TestRecoveryView_ActivatesEveryNonDisabledAction(t *testing.T) { // Base snapshot has no PriorState → event-driven actions land inactive // (pre-BootstrapComplete). Recovery view must clone them with active=