From 74ed12d6b3b986dd9076c43a31b879717412607b Mon Sep 17 00:00:00 2001 From: kai-mathpix Date: Sat, 17 Oct 2026 18:24:35 -0400 Subject: [PATCH] s3 lifecycle: replay only the rules that fit the retention window (#11789) * s3 lifecycle: replay only the rules that fit the retention window runShard handed drainShardEvents the full snapshot, so rules the retention window assigns to the walk were also matched during meta-log replay. Their matches are not due yet, and drainShardEvents freezes the cursor at the first event with a not-yet-due match. With a 3-day and a 30-day expiration rule on one bucket and RetentionWindow of 8 days, every shard's cursor stopped at the first 30-day object's PUT and stayed there until that object came due. Each pass then re-read the log from that point and re-dispatched every 3-day expiration already handled, which surfaces as NOOP_RESOLVED dispatches, a growing cursor_lag_max, and passes that still report status=ok. Snapshot.ReplayView drops the replay-kind actions promoted to the walk and keeps every other action as compiled, and the drain uses it. Promoted rules are served by the walk as before. RulesForShard, PromotedHash and ReplayView share one isPromoted predicate. With RetentionWindow unset no active rule is promoted, so behaviour without a configured window is unchanged. Co-Authored-By: Claude Opus 5.5 * s3 lifecycle: scale the replay-view test with DaysToDuration The test took its retention window from DaysToDuration but dated its PUTs in hours, so under the s3tests build tag (one lifecycle day = 10s) every event fell before the seeded cursor and replay never ran. Express the event ages in DaysToDuration too, so the test exercises replay at both scales. Co-Authored-By: Claude Opus 5.5 * s3 lifecycle: rehabilitate scan_only mode in the replay drain view ReplayView shared the base CompiledAction pointers, so a replay-kind rule locked into ModeScanOnly by a prior compile stayed invisible to router.Route's event_driven gate even when the runtime retention window kept it in the replay partition - and it is absent from the walk view too, so the rule would be served by neither. Clone non-promoted replay-kind actions the way RulesForShard does (Mode forced to event_driven), reactivating the scan_only case whose bootstrap rendezvous already completed, while preserving the base active bit for genuinely pre-bootstrap actions. Disabled actions are excluded for parity with the other views, and DESIGN.md now documents the view. * s3 lifecycle: keep drain activation observable, require walker with retention ReplayView now shares pointers for event-driven actions so a mid-pass MarkActive flip remains visible to the drain; only a scan_only-locked action is cloned to event_driven, and reactivated only when its bootstrap rendezvous completed (tracked via a new bootstrapComplete field on CompiledAction). validate() rejects RetentionWindow without Walker: promoted rules would lose their only dispatch path while the drain cursor still advanced past their events. --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Chris Lu --- weed/s3api/s3lifecycle/DESIGN.md | 5 +- .../s3lifecycle/dailyrun/replay_view_test.go | 67 +++++++++++++++++++ weed/s3api/s3lifecycle/dailyrun/run.go | 11 ++- .../s3lifecycle/dailyrun/run_bounded_test.go | 13 ++++ weed/s3api/s3lifecycle/engine/compile.go | 1 + weed/s3api/s3lifecycle/engine/engine.go | 1 + weed/s3api/s3lifecycle/engine/hashes.go | 6 +- weed/s3api/s3lifecycle/engine/views.go | 60 ++++++++++++++--- weed/s3api/s3lifecycle/engine/views_test.go | 64 ++++++++++++++++++ 9 files changed, 213 insertions(+), 15 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dailyrun/replay_view_test.go 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=