feat(s3/lifecycle): throttle steady-state walker by cfg.WalkerInterval (#9484)
* feat(s3/lifecycle): throttle steady-state walker by cfg.WalkerInterval The steady-state and empty-replay walker fired on every dailyrun.Run invocation, which is fine when Run is called at the bucket-walk cadence the operator intends (e.g., once per hour or once per day), but catastrophic when a fast driver like the s3tests CI workflow or the admin worker scheduler invokes Run at multi-second cadence — each tick ran a full subtree scan per shard, crushing the filer. Decouple walker cadence from Run() invocation cadence: persist LastWalkedNs in the per-shard cursor and fire the steady-state / empty-replay walker only when (runNow - LastWalkedNs) >= cfg.WalkerInterval. Cold-start and recovery walker fires (RecoveryView) stay unconditional since those are bounded events that must run when their trigger condition (no cursor, hash mismatch) is met. Recovery walker fires also update LastWalkedNs so the subsequent steady-state pass doesn't double-walk. cfg.WalkerInterval=0 keeps the prior "fire every pass" behavior — the in-repo integration tests and s3tests fast driver continue to work unchanged. Production deployments should set this to the walk cost budget (typically 1h-24h depending on cluster size). Cursor file is back-compat: last_walked_ns is omitempty, so cursor files written before this change decode as LastWalkedNs=0, which walkerDue treats as "never walked steady-state" → walker fires next pass to establish the anchor (same path a cold-start cursor takes). No version bump. Operator surface for WalkerInterval is the dailyrun.Config struct; plumbing through worker.tasks.s3_lifecycle.Config and the admin schema is a follow-up. * fix(s3/lifecycle): suppress walker double-fire within a single pass Two gemini-code-assist findings: 1. walkerDue with interval=0 returned true even when lastWalkedNs == runNow.UnixNano() — the cold-start / recovery branch already fired the walker this pass, and the steady-state fall-through fired it again. RecoveryView is a superset of every per-shard partition, so the second walk added zero coverage and burned a full subtree scan. Add a within-pass guard at the front of walkerDue: if the cursor's LastWalkedNs equals runNow's UnixNano, the walker already ran this pass — skip. 2. The empty-replay branch passed persisted.LastWalkedNs to walkerDue instead of the local lastWalkedNs variable the rest of runShard threads through. Trivially equal at this point in the function, but the inconsistency would mask a future bug if any code above the branch ever sets lastWalkedNs. Test updates: TestWalkerDue gains the within-pass guard case plus a companion "earlier same pass still fires" sanity check. TestRunShard_ColdStartDoesNotDoubleWalk is new and pins the integration: cold-start runShard with WalkerInterval=0 must call cfg.Walker exactly once, not twice. * fix(s3/lifecycle): reject negative WalkerInterval + lift within-pass guard Two coderabbit findings: 1. validate() now rejects negative cfg.WalkerInterval. A typo like -1h previously fell through walkerDue's `interval <= 0` branch and silently re-enabled "walk every pass" — the exact behavior the throttle was added to prevent. The admin-config parser already clamps negative input to zero, but callers using dailyrun.Config directly (tests, embedders) now get a loud error instead. 2. Within-pass double-fire suppression moves out of walkerDue and into runShard's walkedThisPass local flag. walkerDue's equality check (lastWalkedNs == runNow.UnixNano) was correct in production (each pass freezes runNow at time.Now().UTC, no collisions) but fragile in tests that inject the same runNow across distinct passes — the test would see false suppression. Separating the concerns also makes walkerDue answer one question (persisted-state throttle) and runShard another (within-pass call-site dedup). walker_interval_test.go: TestValidate_RejectsNegativeWalkerInterval pins the new validation. TestWalkerDue's within-pass cases move out (the function is pure throttle now); TestRunShard_ColdStartDoesNot DoubleWalk still pins the integration behavior end-to-end.
This commit is contained in:
Chris Lu
GitHub
parent
75c807b586
commit
c6582228b8
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
276
weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go
Normal file
276
weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user
