diff --git a/weed/s3api/s3lifecycle/dailyrun/cursor.go b/weed/s3api/s3lifecycle/dailyrun/cursor.go index 95b24eb53..fc73a4647 100644 --- a/weed/s3api/s3lifecycle/dailyrun/cursor.go +++ b/weed/s3api/s3lifecycle/dailyrun/cursor.go @@ -39,7 +39,12 @@ const cursorFileVersion = 1 // meta-log event whose Matches all dispatched successfully (or as // NOOP_RESOLVED); RuleSetHash is the content hash of the replay-eligible // rules from the run that wrote this cursor; PromotedHash is the hash -// of replay-eligible rules currently in scan_only. +// of replay-eligible rules currently in scan_only. LastWalkedNs is the +// runNow of the most recent successful steady-state / empty-replay +// walker fire on this shard — runShard uses it together with +// cfg.WalkerInterval to throttle the walker independently of Run() +// invocation frequency (so a worker driven by a 2s test ticker doesn't +// crush filer with a full bucket walk per tick). // // In Phase 2 PromotedHash is always the empty hash because the walker // path isn't wired yet — any rule that would land in walk causes the @@ -48,16 +53,24 @@ type Cursor struct { TsNs int64 RuleSetHash [32]byte PromotedHash [32]byte + LastWalkedNs int64 } // cursorFile is the on-disk JSON shape. Bytes are base64-encoded by // encoding/json automatically. +// +// last_walked_ns is omitempty so cursor files written before the field +// existed still decode cleanly: an older file has no last_walked_ns, +// which marshals back into LastWalkedNs=0, which runShard treats as +// "never walked steady-state" and so the next pass fires the walker +// (the same path a cold-start cursor takes). No version bump needed. type cursorFile struct { Version int `json:"version"` ShardID int `json:"shard_id"` TsNs int64 `json:"ts_ns"` RuleSetHash []byte `json:"rule_set_hash"` PromotedHash []byte `json:"promoted_hash"` + LastWalkedNs int64 `json:"last_walked_ns,omitempty"` } // CursorPersister loads and saves daily-replay cursors. The Phase 2 @@ -119,7 +132,7 @@ func (p *FilerCursorPersister) Load(ctx context.Context, shardID int) (Cursor, b if len(cf.PromotedHash) != 32 { return Cursor{}, false, fmt.Errorf("cursor promoted_hash shard=%d: got %d bytes, want 32", shardID, len(cf.PromotedHash)) } - c := Cursor{TsNs: cf.TsNs} + c := Cursor{TsNs: cf.TsNs, LastWalkedNs: cf.LastWalkedNs} copy(c.RuleSetHash[:], cf.RuleSetHash) copy(c.PromotedHash[:], cf.PromotedHash) return c, true, nil @@ -135,6 +148,7 @@ func (p *FilerCursorPersister) Save(ctx context.Context, shardID int, c Cursor) TsNs: c.TsNs, RuleSetHash: c.RuleSetHash[:], PromotedHash: c.PromotedHash[:], + LastWalkedNs: c.LastWalkedNs, } var buf bytes.Buffer enc := json.NewEncoder(&buf) diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 624ac547d..61bd6496b 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -62,6 +62,23 @@ type Config struct { // still rewinds, matching Phase 4a behavior. Walker WalkerFunc + // WalkerInterval throttles the steady-state and empty-replay walker + // invocations: the walker fires only when the time since the + // per-shard cursor's LastWalkedNs exceeds this value. Cold-start + // and recovery walker fires (RecoveryView) are unconditional — + // those are bounded events that must run once when the trigger + // condition is met. 0 means "fire on every pass" (the prior + // behavior, suitable for s3tests-style rapid-cadence drivers and + // for in-repo integration tests). + // + // Production deployments should set this to roughly the walk cost + // budget per shard per cluster: e.g., 1h for a small cluster, 6h + // for a large one. The walker reads the whole bucket subtree it + // covers, so a too-short interval crushes the filer; a too-long + // interval delays ExpirationDate and ExpiredObjectDeleteMarker + // dispatches. + WalkerInterval time.Duration + ClientName string // 0 -> randomized per-run. ClientID int32 @@ -313,6 +330,16 @@ func validate(cfg Config) error { if cfg.BucketsPath == "" { return errors.New("daily_run: empty BucketsPath") } + // A negative WalkerInterval would silently fall through walkerDue's + // `interval <= 0` branch and re-enable "walk every pass", defeating + // the throttle whose whole point is bounding filer load. The + // admin-config parser already clamps negative input to zero + // (worker/tasks/s3_lifecycle/config.go), but callers using + // dailyrun.Config directly (tests, embedders, future drivers) get + // a loud failure instead. + if cfg.WalkerInterval < 0 { + return fmt.Errorf("daily_run: negative WalkerInterval %v (0 = unthrottled, positive values throttle)", cfg.WalkerInterval) + } 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) @@ -322,12 +349,16 @@ func validate(cfg Config) error { } // runShard executes one daily-replay pass; see DESIGN.md for algorithm. -// Two walker invocations under cfg.Walker (when set): -// - recovery branch: RecoveryView, so already-due objects across the -// rewritten rule set fire before the cursor rewinds. -// - steady state: RulesForShard's walk view, so walker-bound and -// scan_only-promoted rules fire every day even when replay rules -// are unchanged. +// Three walker invocation paths under cfg.Walker (when set): +// - cold start / recovery: RecoveryView. Unconditional — these are +// bounded events (first run for a shard, or rule-content edit / partition +// flip) and the walker must run once when the trigger condition is met. +// - steady state: RulesForShard's walk view, throttled by cfg.WalkerInterval +// against persisted.LastWalkedNs so the walker fires on its own cadence +// independently of how often Run() is invoked. +// - empty replay: same throttling — buckets with only walker-bound rules +// use the walk view, and a too-fast Run() driver would otherwise burn +// the filer with a full subtree scan per tick. func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int, events <-chan *reader.Event) error { shardLabel := strconv.Itoa(shardID) shardStart := time.Now() @@ -361,25 +392,46 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim } promoted := engine.PromotedHash(snap, retentionWindow) + // lastWalkedNs threads the steady-state walker's clock through the + // cursor saves below. Start with the persisted value; only update + // it when a steady-state / empty-replay walker fire actually + // happens. Cold-start and recovery walker fires also bump it so + // the next pass's throttle has a fresh anchor. + // + // walkedThisPass is the in-pass guard: walkerDue answers the + // persisted-state question ("has enough time elapsed?"), but a + // cold-start or recovery branch above the steady-state check + // already fired the walker with RecoveryView, which is a superset + // of every per-shard partition. Keeping the within-pass guard out + // of walkerDue lets that function stay pure (interval/never-walked + // logic only) and side-steps the test-injectable-runNow ambiguity + // where two distinct passes happen to share a UnixNano. + lastWalkedNs := persisted.LastWalkedNs + walkedThisPass := false + if rsh == [32]byte{} { // No replay-eligible rules. Walker-only rules // (ExpirationDate / ExpiredDeleteMarker / NewerNoncurrent) // or scan_only-promoted rules might still need a walk; run - // the steady-state walker before persisting the empty - // cursor and returning. Without this, a bucket whose only - // rule is walker-bound would never have it dispatched — + // the walker (throttled by WalkerInterval) before persisting + // the empty cursor and returning. Without this, a bucket whose + // only rule is walker-bound would never have it dispatched — // the bug TestLifecycleExpirationDateInThePast caught. - if cfg.Walker != nil { + if cfg.Walker != nil && walkerDue(lastWalkedNs, runNow, cfg.WalkerInterval) { if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 { if werr := cfg.Walker(ctx, walkView, shardID); werr != nil { return fmt.Errorf("shard=%d: steady walk (empty replay): %w", shardID, werr) } + lastWalkedNs = runNow.UnixNano() + walkedThisPass = true } } + _ = walkedThisPass // empty-replay branch returns; no downstream walker. return cfg.Persister.Save(ctx, shardID, Cursor{ TsNs: 0, RuleSetHash: rsh, PromotedHash: promoted, + LastWalkedNs: lastWalkedNs, }) } @@ -399,6 +451,8 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim if werr := cfg.Walker(ctx, engine.RecoveryView(snap), shardID); werr != nil { return fmt.Errorf("shard=%d: recovery walk: %w", shardID, werr) } + lastWalkedNs = runNow.UnixNano() + walkedThisPass = true } if mustWalkRecovery { // Rule changed: rewind cursor so the sliding replay re-scans @@ -408,6 +462,7 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim TsNs: runNow.Add(-maxTTL).UnixNano(), RuleSetHash: rsh, PromotedHash: promoted, + LastWalkedNs: lastWalkedNs, } return cfg.Persister.Save(ctx, shardID, next) } @@ -421,11 +476,17 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim // hash already accounted for. Empty walk view (no rules need walking // today) skips the call so non-versioned, replay-only deployments // don't pay an O(N) bucket-walk per run. - if cfg.Walker != nil { + // + // Two-stage gate: walkedThisPass suppresses a double walk when the + // cold-start branch above already fired (RecoveryView covered it). + // walkerDue applies the persisted-state throttle for the case + // where this is a normal pass that didn't trigger cold-start. + if cfg.Walker != nil && !walkedThisPass && walkerDue(lastWalkedNs, runNow, cfg.WalkerInterval) { if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 { if werr := cfg.Walker(ctx, walkView, shardID); werr != nil { return fmt.Errorf("shard=%d: steady walk: %w", shardID, werr) } + lastWalkedNs = runNow.UnixNano() } } @@ -452,7 +513,7 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim saveCtx, saveCancel := context.WithTimeout(context.Background(), 5*time.Second) defer saveCancel() if drainErr != nil { - _ = cfg.Persister.Save(saveCtx, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted}) + _ = cfg.Persister.Save(saveCtx, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted, LastWalkedNs: lastWalkedNs}) // passCtx timeout is the expected end-of-pass for an idle // subscription; not a real error. Other drain errors still // propagate. @@ -466,9 +527,30 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted, + LastWalkedNs: lastWalkedNs, }) } +// walkerDue answers the persisted-state throttle: has enough time +// elapsed since the last walker fire on this shard? interval == 0 +// means "fire every pass" (the prior, unconditional behavior). +// lastWalkedNs == 0 means "never walked steady-state" — fire to +// establish the throttle anchor. Otherwise gate on +// (runNow - lastWalkedNs) >= interval. +// +// Within-pass double-fire suppression (cold-start / recovery walker +// already ran this pass; steady-state branch must not re-walk a +// superset partition) lives in runShard's walkedThisPass local flag +// rather than here. Keeping walkerDue pure means the function answers +// one question and a test that injects two distinct passes with the +// same runNow doesn't get falsely throttled. +func walkerDue(lastWalkedNs int64, runNow time.Time, interval time.Duration) bool { + if interval <= 0 || lastWalkedNs == 0 { + return true + } + return runNow.UnixNano()-lastWalkedNs >= int64(interval) +} + // drainShardEvents reads pre-fanned-out events for this shard from the // shared meta-log subscription and dispatches matches whose due_time is // past runNow. cursorAdvanceTo stops growing at the first event with a diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go new file mode 100644 index 000000000..26099ee83 --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go @@ -0,0 +1,276 @@ +package dailyrun + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_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/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// readerEventAlias keeps the test free of the reader import collision +// in the recovery test file when we need a typed nil channel. +type readerEventAlias = reader.Event + +// TestWalkerDue covers the throttle decision matrix in isolation. +func TestWalkerDue(t *testing.T) { + runNow := time.Unix(1_700_000_000, 0).UTC() + cases := []struct { + name string + lastWalkedNs int64 + interval time.Duration + want bool + }{ + // interval=0 keeps the prior "fire every pass" semantics — the + // in-repo integration tests and the s3tests fast driver rely on + // this so the rule-change-then-walk behavior surfaces within a + // single test runtime. Within-pass double-fire suppression + // lives in runShard's walkedThisPass local; walkerDue answers + // the persisted-state question only. + {"interval zero always due", runNow.UnixNano(), 0, true}, + // LastWalkedNs=0 marks "never walked steady-state" — the post- + // upgrade cold-start case for cursor files that predate the + // LastWalkedNs field. Fire so the throttle anchor gets seeded. + {"never walked is due", 0, time.Hour, true}, + // Throttle on: interval elapsed → due. + {"interval elapsed", runNow.Add(-2 * time.Hour).UnixNano(), time.Hour, true}, + {"interval exactly elapsed", runNow.Add(-time.Hour).UnixNano(), time.Hour, true}, + // Throttle on: not yet → skip. + {"interval not yet elapsed", runNow.Add(-30 * time.Minute).UnixNano(), time.Hour, false}, + // Future LastWalkedNs (clock skew or replay of an older runNow + // against a newer cursor) must not panic and must not fire — + // treat as "throttled" because the cursor claims a future walk. + {"future lastWalked", runNow.Add(time.Hour).UnixNano(), time.Hour, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, walkerDue(tc.lastWalkedNs, runNow, tc.interval)) + }) + } +} + +// TestRunShard_WalkerThrottle confirms two back-to-back runShard +// invocations only fire the steady-state walker once when the second +// pass starts inside the WalkerInterval window. The interval-zero +// control case keeps firing on every pass — same shape, different +// expectation — pinning the prior behavior so a default-zero config +// (tests, in-repo integration) doesn't regress. +func TestRunShard_WalkerThrottle(t *testing.T) { + cases := []struct { + name string + interval time.Duration + secondPassAfter time.Duration + wantTotalCalls int + wantSecondAdvNs bool // did LastWalkedNs change between pass 1 and pass 2? + }{ + {"interval=0 fires every pass", 0, 30 * time.Second, 2, true}, + {"throttled: second pass within interval", time.Hour, 30 * time.Second, 1, false}, + {"throttled: second pass past interval", time.Hour, 2 * time.Hour, 2, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + snap := snapshotWithRule(t, 30) // replay-eligible only; steady-state walker would normally skip + // Force a walker-only partition so the steady-state branch + // has something to walk: set retentionWindow=0 in cfg, which + // makes promoted non-empty and walkView non-empty. + p := newMemPersister() + calls := 0 + cfg := Config{ + Persister: p, + Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + calls++ + return nil + }, + RetentionWindow: -1, // negative sentinel: falls back to maxTTL — leaves walkView empty + WalkerInterval: tc.interval, + } + // Pre-seed cursor matching snap's hashes so cold-start / + // recovery branches don't fire and we measure ONLY the + // steady-state walker. + rsh := engine.ReplayContentHash(snap) + promoted := engine.PromotedHash(snap, engine.MaxEffectiveTTL(snap)) + runNow := time.Unix(1_700_000_000, 0).UTC() + require.NoError(t, p.Save(context.Background(), 0, Cursor{ + TsNs: runNow.UnixNano(), + RuleSetHash: rsh, + PromotedHash: promoted, + })) + + // With the snapshot's only rule being replay-eligible, the + // steady-state walkView is empty and the walker won't fire + // regardless of throttle. Use a snapshot that has BOTH a + // replay rule AND a scan-only walker rule by forcing + // retentionWindow to a very small value via cfg. + cfg.RetentionWindow = time.Nanosecond + // Recompute promoted with the test retention so persisted + // hashes match what runShard sees on each pass. + snapPromoted := engine.PromotedHash(snap, time.Nanosecond) + require.NoError(t, p.Save(context.Background(), 0, Cursor{ + TsNs: runNow.UnixNano(), + RuleSetHash: rsh, + PromotedHash: snapPromoted, + })) + + // runShard reaches drainShardEvents after the steady-state + // walker fires (replay-eligible rules present). A nil events + // channel would block forever; a closed one returns + // immediately so drain exits cleanly and the cursor save + // still records LastWalkedNs. + closedEvents := make(chan *readerEventAlias) + close(closedEvents) + + // Pass 1. + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, closedEvents)) + afterPass1, _, _ := p.Load(context.Background(), 0) + require.Equal(t, 1, calls, "pass 1 must fire the walker (cold-start anchor)") + require.Equal(t, runNow.UnixNano(), afterPass1.LastWalkedNs) + + // Pass 2. + runNow2 := runNow.Add(tc.secondPassAfter) + closedEvents2 := make(chan *readerEventAlias) + close(closedEvents2) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow2, 0, closedEvents2)) + afterPass2, _, _ := p.Load(context.Background(), 0) + assert.Equal(t, tc.wantTotalCalls, calls, "walker calls after 2 passes") + if tc.wantSecondAdvNs { + assert.Equal(t, runNow2.UnixNano(), afterPass2.LastWalkedNs, "throttle allowed the second walk") + } else { + assert.Equal(t, runNow.UnixNano(), afterPass2.LastWalkedNs, "throttle suppressed the second walk; anchor unchanged") + } + }) + } +} + +// TestValidate_RejectsNegativeWalkerInterval pins the loud-failure +// contract for an embedder or test that constructs dailyrun.Config +// directly with a negative WalkerInterval. The admin-config parser +// clamps negative input to zero (worker/tasks/s3_lifecycle/config.go), +// but a caller bypassing that parser would otherwise get the silent +// fall-through-to-walk-every-pass behavior the throttle is trying to +// prevent. +func TestValidate_RejectsNegativeWalkerInterval(t *testing.T) { + // validate only checks for nil fields, not behavior — type-assert + // each interface to its zero value via a stub. The stubs never run. + cfg := validatableConfig() + cfg.WalkerInterval = -time.Hour + err := validate(cfg) + require.Error(t, err) + assert.Contains(t, err.Error(), "WalkerInterval") + // And the zero sentinel still passes. + cfg.WalkerInterval = 0 + require.NoError(t, validate(cfg)) +} + +// validatableConfig builds a minimal dailyrun.Config that passes the +// nil checks in validate() so individual fields can be poked. The +// stubs intentionally don't implement any meaningful behavior — they +// only need to be non-nil interface values. +func validatableConfig() Config { + return Config{ + Engine: engine.New(), + FilerClient: stubFilerClient{}, + Client: stubLifecycleClient{}, + Persister: newMemPersister(), + Lister: stubSiblingLister{}, + BucketsPath: "/buckets", + } +} + +type stubFilerClient struct{ filer_pb.SeaweedFilerClient } +type stubLifecycleClient struct{} + +func (stubLifecycleClient) LifecycleDelete(_ context.Context, _ *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + return nil, nil +} + +type stubSiblingLister struct{} + +func (stubSiblingLister) Survivors(_ context.Context, _, _ string) (router.Survivors, error) { + return router.Survivors{}, nil +} +func (stubSiblingLister) LookupVersion(_ context.Context, _, _, _ string) (*filer_pb.Entry, error) { + return nil, nil +} +func (stubSiblingLister) ListVersions(_ context.Context, _, _ string) ([]*filer_pb.Entry, error) { + return nil, nil +} +func (stubSiblingLister) LookupNullVersion(_ context.Context, _, _ string) (*filer_pb.Entry, bool, error) { + return nil, false, nil +} + +// TestRunShard_ColdStartDoesNotDoubleWalk pins the within-pass guard +// in runShard's walkedThisPass local. Before the guard, a cold-start +// pass with WalkerInterval=0 fired the walker twice in one pass: +// once via the mustWalkColdStart branch (with RecoveryView) and again +// immediately via the steady-state branch (with the per-shard walk +// view, which is a subset of RecoveryView). The second walk added no +// coverage and burned a full bucket scan. Guard against regression. +func TestRunShard_ColdStartDoesNotDoubleWalk(t *testing.T) { + snap := snapshotWithRule(t, 30) + p := newMemPersister() + // No persisted cursor → mustWalkColdStart=true. WalkerInterval=0 + // would, before the fix, allow the steady-state branch to fire + // again with no time elapsed. + calls := 0 + cfg := Config{ + Persister: p, + Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + calls++ + return nil + }, + WalkerInterval: 0, + } + runNow := time.Unix(1_700_000_000, 0).UTC() + + closedEvents := make(chan *readerEventAlias) + close(closedEvents) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, closedEvents)) + + assert.Equal(t, 1, calls, "cold-start walker must fire exactly once per pass even with interval=0") +} + +// TestRunShard_RecoveryWalkerSetsLastWalkedAnchor pins that the +// unconditional recovery-branch walker fire still updates the +// LastWalkedNs anchor so the NEXT pass's throttle starts counting +// from this walk rather than re-firing immediately. +func TestRunShard_RecoveryWalkerSetsLastWalkedAnchor(t *testing.T) { + snap := snapshotWithRule(t, 30) + p := newMemPersister() + var stale [32]byte + for i := range stale { + stale[i] = 0xAA + } + require.NoError(t, p.Save(context.Background(), 5, Cursor{ + TsNs: 1234, + RuleSetHash: stale, + LastWalkedNs: 0, // never walked anchor + })) + + calls := 0 + cfg := Config{ + Persister: p, + Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + calls++ + return nil + }, + WalkerInterval: time.Hour, + } + runNow := time.Unix(1_700_000_000, 0).UTC() + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 5, nil)) + + require.Equal(t, 1, calls, "recovery branch fires the walker unconditionally") + got, ok, _ := p.Load(context.Background(), 5) + require.True(t, ok) + assert.Equal(t, runNow.UnixNano(), got.LastWalkedNs, + "recovery walk must update LastWalkedNs so the steady-state pass after it can throttle") +} + +// Compile-time check that the documented sentinel passes the type system. +var _ = s3lifecycle.ShardCount