feat(s3/lifecycle): bootstrap re-walk cadence + operator hooks (Phase 8) (#9386)

* feat(s3/lifecycle): bootstrap re-walk cadence + operator hooks (Phase 8)

scan_only actions only fire from the bootstrap walk: the engine
classifies a rule as scan_only when its retention horizon exceeds the
meta-log retention, so event-driven routing can't be trusted. Today
each bucket walks once per process, so a long-running worker never
revisits — scan_only retention only catches up when the worker
restarts.

Replace BucketBootstrapper.known (set) with BucketBootstrapper.lastWalk
(name -> completion time). KickOffNew now re-walks a bucket whose
last walk completed more than BootstrapInterval ago. Zero interval
preserves the legacy walk-once-per-process behavior so existing
deployments don't change cadence by default. walkBucket re-stamps
on success and clears the stamp on failure (via MarkDirty), so the
next KickOffNew picks failed walks back up.

Add MarkDirty / MarkAllDirty operator hooks for forced re-walks, and
a Now func() for testable time travel.

weed shell run-shard grows --bootstrap-interval (cadence knob) and
--force-bootstrap (drop in-memory state at startup so every bucket
walks again immediately, useful when a config change should take
effect without a restart).

Tests: cadence respected (skip inside interval, re-walk past it);
zero interval keeps once-per-process; MarkDirty forces re-walk
under a 24h interval; MarkAllDirty resets every record. The
fakeClock helper guards the test clock with a mutex so race-detector
runs are clean.

* fix(s3/lifecycle): split walk state, thread BootstrapInterval through worker, drop dead flag

Three issues with the Phase 8 cadence work as it landed:

1. lastWalk did double duty as both completed-walk timestamp and
   in-flight debounce. A walk that took longer than BootstrapInterval
   would have a fresh KickOffNew start a duplicate goroutine on the
   next refresh tick because the stamp from KickOffNew looked stale
   against the interval. Split into lastCompleted (set on success)
   and inFlight (set on dispatch, cleared after the walk goroutine
   returns success or failure). KickOffNew skips inFlight buckets
   regardless of cadence.

2. The cadence knob existed on `weed shell` but not on the production
   path: scheduler.Scheduler constructed BucketBootstrapper without
   BootstrapInterval, and weed/worker/tasks/s3_lifecycle/Config had
   no field for it. Add Scheduler.BootstrapInterval, parse
   `bootstrap_interval_minutes` in ParseConfig (zero = legacy walk-
   once-per-process; negative clamps to zero), and forward it from
   the handler. Tests cover default, override, clamp, and explicit-zero.

3. --force-bootstrap was a no-op: BucketBootstrapper is freshly
   allocated at command start, so MarkAllDirty on empty state does
   nothing, and the flag couldn't influence an already-running
   process anyway. Remove it; a real runtime trigger (SIGHUP, control
   RPC) is a separate change.

In-flight regression: a blockingInjector pins the first walk in
progress while the test advances the clock past the interval. The
second KickOffNew is a no-op (inFlight check). After release, the
post-completion KickOffNew within the interval is also a no-op.

* test(s3/lifecycle): wait for lastCompleted stamp before advancing fake clock

The cadence test polled listedN to know "the walk happened" — but
that fires once both list passes are issued, while the success-stamp
lands later, after walkBucketDir returns. A clock.Advance(30m)
between those two events would record the stamp at clock+30m
instead of T0; the next assertion would then see now.Sub(last) < 1h
and skip the expected re-walk. Tight in practice but exposed under
-race / load.

Add a waitForCompleted helper that polls b.lastCompleted directly,
and use it before each clock advance in both the cadence and zero-
interval tests.

* fix(s3/lifecycle): expose bootstrap interval in worker UI; honor MarkDirty during walks

Two follow-ups on Phase 8.

The worker config descriptor had no bootstrap_interval_minutes field,
so the production operator UI couldn't enable the cadence — only the
internal ParseConfig + Scheduler wiring knew about it. Add the field
to the cadence section (MinValue=0 since 0 is the legacy default) and
include the default in DefaultValues so existing deployments see the
knob with the right preset.

MarkDirty / MarkAllDirty silently lost their effect when a walk was
in flight: the methods cleared lastCompleted, but the walk's success
path then wrote a fresh timestamp, hiding the operator's invalidation.
Track a pendingDirty set; the walk goroutine consumes the flag on
exit and skips the success stamp, so the next KickOffNew picks the
bucket up immediately.

Regression: pin a walk in progress with a blockingInjector, MarkDirty
the bucket, release the walk, and assert lastCompleted stayed empty
plus the next KickOffNew triggers a new walk inside the
BootstrapInterval window.

* refactor(s3/lifecycle): drop unused MarkDirty / MarkAllDirty + pendingDirty

These methods were the operator-hook half of Phase 8, but the only
caller (--force-bootstrap on the shell command) was removed when it
turned out to be a no-op against a freshly-allocated bootstrapper.
Nothing in production calls them anymore.

Strip the dead surface: MarkDirty, MarkAllDirty, the pendingDirty
set, the dirty-suppression branch in walkBucket, and the three tests
that only exercised those methods. BootstrapInterval-driven
re-bootstrap is the live mechanism. A real runtime trigger (SIGHUP,
control RPC) is a separate change with a real call site.

weed/s3api/s3lifecycle/scheduler/bootstrap.go

@@ -78,33 +78,70 @@ func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string
 // Synthesized events carry TsNs=0 so dispatcher.advance is a no-op for
 // them — the reader still resumes from its persisted cursor on restart.
 //
-// Bucket completion is in-memory per process; a fresh worker run walks
-// each bucket once on first refresh.
+// Walk state is tracked in-memory per process via two maps:
+//   - lastCompleted: time of the last successful walk, drives the
+//     BootstrapInterval cadence; zero interval means "walk once per
+//     process".
+//   - inFlight: buckets whose walk goroutines are still running.
+//     Skipped regardless of cadence so a walk that takes longer than
+//     BootstrapInterval can't trigger a duplicate goroutine on the
+//     next refresh.
 type BucketBootstrapper struct {
 	FilerClient filer_pb.SeaweedFilerClient
 	BucketsPath string
 	Injector    EventInjector
 
-	mu    sync.Mutex
-	known map[string]bool
+	// BootstrapInterval gates re-walks. Zero means "walk once per
+	// process". Non-zero means "walk again once it's been at least
+	// this long since the last completed walk" — the cadence scan_only
+	// actions rely on, since they can only fire from bootstrap.
+	BootstrapInterval time.Duration
+
+	// Now overrides time.Now for tests.
+	Now func() time.Time
+
+	mu            sync.Mutex
+	lastCompleted map[string]time.Time
+	inFlight      map[string]bool
+}
+
+func (b *BucketBootstrapper) now() time.Time {
+	if b.Now != nil {
+		return b.Now()
+	}
+	return time.Now()
 }
 
 // KickOffNew launches a one-shot walker goroutine for every bucket
-// in `buckets` that hasn't been seen before.
+// that's not currently in flight and either has never completed a walk
+// or whose last successful walk finished more than BootstrapInterval ago.
 func (b *BucketBootstrapper) KickOffNew(ctx context.Context, buckets []string) {
 	if b.Injector == nil {
 		return
 	}
+	now := b.now()
 	b.mu.Lock()
-	if b.known == nil {
-		b.known = map[string]bool{}
+	if b.lastCompleted == nil {
+		b.lastCompleted = map[string]time.Time{}
+	}
+	if b.inFlight == nil {
+		b.inFlight = map[string]bool{}
 	}
 	fresh := make([]string, 0, len(buckets))
 	for _, bucket := range buckets {
-		if b.known[bucket] {
+		if b.inFlight[bucket] {
+			// Walk still running from an earlier KickOffNew — never
+			// double up regardless of cadence. A large bucket that
+			// takes longer than BootstrapInterval would otherwise
+			// have a fresh goroutine fire on every refresh tick.
 			continue
 		}
-		b.known[bucket] = true
+		if last, ok := b.lastCompleted[bucket]; ok {
+			if b.BootstrapInterval <= 0 || now.Sub(last) < b.BootstrapInterval {
+				continue
+			}
+		}
+		b.inFlight[bucket] = true
 		fresh = append(fresh, bucket)
 	}
 	b.mu.Unlock()
@@ -152,9 +189,22 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 		count++
 		return b.Injector.InjectEvent(ctx, ev)
 	}
-	if err := walkBucketDir(ctx, b.FilerClient, root, root, cb); err != nil {
+	walkErr := walkBucketDir(ctx, b.FilerClient, root, root, cb)
+	b.mu.Lock()
+	delete(b.inFlight, bucket)
+	if walkErr == nil {
+		// Stamp completion so BootstrapInterval cadence measures from
+		// end-of-walk. Failures leave lastCompleted alone, so the next
+		// KickOffNew sees no record and walks the bucket again.
+		if b.lastCompleted == nil {
+			b.lastCompleted = map[string]time.Time{}
+		}
+		b.lastCompleted[bucket] = b.now()
+	}
+	b.mu.Unlock()
+	if walkErr != nil {
 		if ctx.Err() == nil {
-			glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, err)
+			glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, walkErr)
 		}
 		return
 	}

weed/s3api/s3lifecycle/scheduler/bootstrap_test.go

@@ -396,9 +396,11 @@ func TestBucketBootstrapper_KickOffNew_LaunchesPerBucket(t *testing.T) {
 
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	assert.True(t, b.known["bucketA"])
-	assert.True(t, b.known["bucketB"])
-	assert.Len(t, b.known, 2)
+	_, hasA := b.lastCompleted["bucketA"]
+	_, hasB := b.lastCompleted["bucketB"]
+	assert.True(t, hasA)
+	assert.True(t, hasB)
+	assert.Len(t, b.lastCompleted, 2)
 }
 
 func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) {
@@ -442,8 +444,9 @@ func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) {
 
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	assert.Len(t, b.known, 3)
-	assert.True(t, b.known["bucketC"])
+	assert.Len(t, b.lastCompleted, 3)
+	_, hasC := b.lastCompleted["bucketC"]
+	assert.True(t, hasC)
 }
 
 func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) {
@@ -456,12 +459,13 @@ func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) {
 	require.NotPanics(t, func() {
 		b.KickOffNew(context.Background(), []string{"bucketA"})
 	})
-	// No walks must have been kicked off, and known must remain empty.
+	// No walks must have been kicked off, and the lastWalk map must
+	// remain empty.
 	time.Sleep(20 * time.Millisecond)
 	assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN))
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	assert.Empty(t, b.known)
+	assert.Empty(t, b.lastCompleted)
 }
 
 func TestBucketBootstrapper_KickOffNew_EmptyBucketListIsNoop(t *testing.T) {
@@ -1100,3 +1104,193 @@ func TestWalkBucketDir_PaginatesBeyondListingLimit(t *testing.T) {
 	// entries at page 2 = 3 paginated calls per pass = 6 total.
 	assert.Equal(t, 6, calls)
 }
+
+// fakeClock is a thread-safe time source for tests that need to fast-
+// forward across a BootstrapInterval boundary. Bootstrap goroutines
+// read it concurrently with the test advancing it, so a plain
+// `clock := time.Now()` plus closure write would race under -race.
+type fakeClock struct {
+	mu sync.Mutex
+	t  time.Time
+}
+
+func newFakeClock() *fakeClock { return &fakeClock{t: time.Now()} }
+
+func (c *fakeClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.t
+}
+
+func (c *fakeClock) Advance(d time.Duration) {
+	c.mu.Lock()
+	c.t = c.t.Add(d)
+	c.mu.Unlock()
+}
+
+// waitForCompleted blocks until the bootstrapper has stamped a
+// lastCompleted entry for the given bucket. Polling listedN is not
+// enough — that fires once both list passes have started, but
+// walkBucket stamps lastCompleted only after walkBucketDir returns,
+// so a clock.Advance between those events would record the stamp at
+// post-advance time and skew BootstrapInterval cadence assertions.
+func waitForCompleted(t *testing.T, b *BucketBootstrapper, bucket string) {
+	t.Helper()
+	waitFor(t, func() bool {
+		b.mu.Lock()
+		_, ok := b.lastCompleted[bucket]
+		b.mu.Unlock()
+		return ok
+	}, "lastCompleted stamp for "+bucket)
+}
+
+func TestBucketBootstrapper_KickOffNew_BootstrapIntervalRevisitsBucket(t *testing.T) {
+	// scan_only actions only fire from bootstrap, so a long-running
+	// worker has to revisit each bucket on a cadence. With
+	// BootstrapInterval set, KickOffNew re-walks once enough wall-clock
+	// has passed since the last completed walk.
+	client := newEmptyFilerClient()
+	inj := &recordingInjector{}
+	clock := newFakeClock()
+	b := &BucketBootstrapper{
+		FilerClient:       client,
+		BucketsPath:       "/buckets",
+		Injector:          inj,
+		BootstrapInterval: time.Hour,
+		Now:               clock.Now,
+	}
+
+	// First wave: walks once. Wait for the goroutine to actually stamp
+	// lastCompleted before advancing the clock — otherwise the stamp
+	// could land at clock+30m instead of T0 and the cadence assertion
+	// would race.
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	waitForCompleted(t, b, "bucketA")
+	firstCount := atomic.LoadInt32(&client.listedN)
+
+	// Inside the interval: skip.
+	clock.Advance(30 * time.Minute)
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	time.Sleep(20 * time.Millisecond)
+	if got := atomic.LoadInt32(&client.listedN); got != firstCount {
+		t.Fatalf("inside interval, must not re-walk; listedN=%d, want %d", got, firstCount)
+	}
+
+	// Past the interval: re-walk.
+	clock.Advance(45 * time.Minute) // total elapsed > 1h
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	waitFor(t, func() bool { return atomic.LoadInt32(&client.listedN) >= firstCount+2 }, "re-walk after interval")
+}
+
+func TestBucketBootstrapper_KickOffNew_ZeroIntervalLegacyOnceOnly(t *testing.T) {
+	// BootstrapInterval == 0 preserves the original "walk once per
+	// process" behavior so existing deployments don't get a different
+	// cadence by default.
+	client := newEmptyFilerClient()
+	inj := &recordingInjector{}
+	clock := newFakeClock()
+	b := &BucketBootstrapper{
+		FilerClient: client,
+		BucketsPath: "/buckets",
+		Injector:    inj,
+		Now:         clock.Now,
+	}
+
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	waitForCompleted(t, b, "bucketA")
+	firstCount := atomic.LoadInt32(&client.listedN)
+
+	// Even after 100 hours, KickOffNew skips the bucket.
+	clock.Advance(100 * time.Hour)
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	time.Sleep(20 * time.Millisecond)
+	if got := atomic.LoadInt32(&client.listedN); got != firstCount {
+		t.Fatalf("zero interval must keep the once-per-process behavior; listedN=%d, want %d", got, firstCount)
+	}
+}
+
+// blockingInjector lets the test pin a walk in progress until it
+// signals release. Useful for asserting in-flight debounce.
+type blockingInjector struct {
+	mu       sync.Mutex
+	events   []*reader.Event
+	released chan struct{}
+}
+
+func newBlockingInjector() *blockingInjector {
+	return &blockingInjector{released: make(chan struct{})}
+}
+
+func (b *blockingInjector) InjectEvent(ctx context.Context, ev *reader.Event) error {
+	select {
+	case <-b.released:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	b.mu.Lock()
+	b.events = append(b.events, ev)
+	b.mu.Unlock()
+	return nil
+}
+
+func (b *blockingInjector) release() { close(b.released) }
+
+func (b *blockingInjector) eventCount() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return len(b.events)
+}
+
+func TestBucketBootstrapper_KickOffNew_InFlightDebounceBlocksDuplicate(t *testing.T) {
+	// A walk that takes longer than BootstrapInterval would otherwise
+	// have a fresh KickOffNew start a duplicate goroutine on the next
+	// refresh tick. The inFlight set prevents that. Verify by:
+	//  1) Pinning a walk in progress via a blockingInjector,
+	//  2) Advancing the clock past BootstrapInterval,
+	//  3) Confirming the second KickOffNew is a no-op while the first
+	//     is still running,
+	//  4) Releasing the first walk and asserting only one walk
+	//     completed (one bucket-root listing pair).
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			"/buckets/bucketA": {fileEntry("a.txt")},
+		},
+	}
+	inj := newBlockingInjector()
+	clock := newFakeClock()
+	b := &BucketBootstrapper{
+		FilerClient:       client,
+		BucketsPath:       "/buckets",
+		Injector:          inj,
+		BootstrapInterval: time.Hour,
+		Now:               clock.Now,
+	}
+
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	// Wait for the walk goroutine to actually start listing (pass 1
+	// fires a ListEntries before InjectEvent).
+	waitFor(t, func() bool { return atomic.LoadInt32(&client.listedN) >= 1 }, "first walk to begin listing")
+
+	// Advance past the interval — a stale-state KickOffNew would now
+	// see the lastCompleted as expired and try again.
+	clock.Advance(2 * time.Hour)
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	time.Sleep(20 * time.Millisecond)
+	if got := inj.eventCount(); got != 0 {
+		t.Fatalf("first walk still blocked, got %d injected events from a phantom second walk", got)
+	}
+
+	// Release: the first walk completes. eventCount goes to 1.
+	inj.release()
+	waitFor(t, func() bool { return inj.eventCount() == 1 }, "first walk to drain")
+	// Even after another KickOffNew at the same simulated time, the
+	// in-flight is now cleared and lastCompleted is fresh — second
+	// KickOffNew within interval is a no-op.
+	prevListed := atomic.LoadInt32(&client.listedN)
+	b.KickOffNew(context.Background(), []string{"bucketA"})
+	time.Sleep(20 * time.Millisecond)
+	if got := atomic.LoadInt32(&client.listedN); got != prevListed {
+		t.Fatalf("post-release KickOffNew within interval must be a no-op; listedN=%d, want %d", got, prevListed)
+	}
+}
+

weed/s3api/s3lifecycle/scheduler/scheduler.go

@@ -33,11 +33,12 @@ type Scheduler struct {
 	ClientID    int32
 	ClientName  string
 
-	Workers         int
-	DispatchTick    time.Duration
-	CheckpointTick  time.Duration
-	RefreshInterval time.Duration
-	RetryBackoff    time.Duration
+	Workers           int
+	DispatchTick      time.Duration
+	CheckpointTick    time.Duration
+	RefreshInterval   time.Duration
+	BootstrapInterval time.Duration
+	RetryBackoff      time.Duration
 }
 
 // Run blocks until ctx is canceled. Spawns Workers + 1 goroutines: one
@@ -100,9 +101,10 @@ func (s *Scheduler) Run(ctx context.Context) error {
 	}
 
 	bs := &BucketBootstrapper{
-		FilerClient: s.FilerClient,
-		BucketsPath: s.BucketsPath,
-		Injector:    pipelineFanout(pipelinesByShard),
+		FilerClient:       s.FilerClient,
+		BucketsPath:       s.BucketsPath,
+		Injector:          pipelineFanout(pipelinesByShard),
+		BootstrapInterval: s.BootstrapInterval,
 	}
 
 	s.refreshEngine(ctx, bs)

weed/shell/command_s3_lifecycle_run_shard.go

@@ -68,6 +68,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i
 	dispatchTick := fs.Duration("dispatch", 5*time.Second, "dispatcher tick cadence")
 	checkpointTick := fs.Duration("checkpoint", 30*time.Second, "cursor checkpoint cadence")
 	refreshInterval := fs.Duration("refresh", 5*time.Minute, "interval for rebuilding the engine snapshot from filer-backed bucket configs; 0 = compile once at startup")
+	bootstrapInterval := fs.Duration("bootstrap-interval", 0, "cadence for revisiting each bucket's bootstrap walk; 0 = walk once per process. scan_only actions only fire from bootstrap, so a long-running worker needs a non-zero value to handle their retention horizon")
 	runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout. -events alone can hang on quiet shards")
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -119,9 +120,10 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i
 		}
 
 		bsr := &scheduler.BucketBootstrapper{
-			FilerClient: filerClient,
-			BucketsPath: bucketsPath,
-			Injector:    pipeline,
+			FilerClient:       filerClient,
+			BucketsPath:       bucketsPath,
+			Injector:          pipeline,
+			BootstrapInterval: *bootstrapInterval,
 		}
 		bootstrapCtx, bootstrapCancel := context.WithCancel(context.Background())
 		defer bootstrapCancel()

weed/worker/tasks/s3_lifecycle/config.go

@@ -9,31 +9,34 @@ import (
 const (
 	jobType = "s3_lifecycle"
 
-	defaultWorkers                = 1
-	defaultDispatchTickMinutes    = int64(1)
-	defaultCheckpointTickSeconds  = int64(30)
-	defaultRefreshIntervalMinutes = int64(5)
-	defaultMaxRuntimeMinutes      = int64(60)
+	defaultWorkers                  = 1
+	defaultDispatchTickMinutes      = int64(1)
+	defaultCheckpointTickSeconds    = int64(30)
+	defaultRefreshIntervalMinutes   = int64(5)
+	defaultMaxRuntimeMinutes        = int64(60)
+	defaultBootstrapIntervalMinutes = int64(0) // 0 = walk once per process
 )
 
 // Config is the parsed AdminConfigForm + WorkerConfigForm view.
 type Config struct {
-	Workers         int
-	DispatchTick    time.Duration
-	CheckpointTick  time.Duration
-	RefreshInterval time.Duration
-	MaxRuntime      time.Duration
+	Workers           int
+	DispatchTick      time.Duration
+	CheckpointTick    time.Duration
+	RefreshInterval   time.Duration
+	BootstrapInterval time.Duration
+	MaxRuntime        time.Duration
 }
 
 // ParseConfig pulls the lifecycle Handler config from the merged
 // admin+worker config values. Missing fields fall back to defaults.
 func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Config {
 	cfg := Config{
-		Workers:         int(readInt64(adminValues, "workers", defaultWorkers)),
-		DispatchTick:    time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute,
-		CheckpointTick:  time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second,
-		RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute,
-		MaxRuntime:      time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute,
+		Workers:           int(readInt64(adminValues, "workers", defaultWorkers)),
+		DispatchTick:      time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute,
+		CheckpointTick:    time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second,
+		RefreshInterval:   time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute,
+		BootstrapInterval: time.Duration(readInt64(workerValues, "bootstrap_interval_minutes", defaultBootstrapIntervalMinutes)) * time.Minute,
+		MaxRuntime:        time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute,
 	}
 	if cfg.Workers <= 0 {
 		cfg.Workers = defaultWorkers
@@ -47,6 +50,13 @@ func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Co
 	if cfg.RefreshInterval <= 0 {
 		cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute
 	}
+	// BootstrapInterval is intentionally NOT clamped — zero means
+	// "walk once per process", which is the legacy default for any
+	// deployment that hasn't opted into a cadence yet. Negative values
+	// fall back to zero.
+	if cfg.BootstrapInterval < 0 {
+		cfg.BootstrapInterval = 0
+	}
 	if cfg.MaxRuntime <= 0 {
 		cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute
 	}

weed/worker/tasks/s3_lifecycle/config_test.go

@@ -24,6 +24,9 @@ func TestParseConfigDefaults(t *testing.T) {
 	if cfg.MaxRuntime != 60*time.Minute {
 		t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime)
 	}
+	if cfg.BootstrapInterval != 0 {
+		t.Errorf("BootstrapInterval default=%v, want 0 (walk-once-per-process)", cfg.BootstrapInterval)
+	}
 }
 
 func TestParseConfigOverrides(t *testing.T) {
@@ -31,10 +34,11 @@ func TestParseConfigOverrides(t *testing.T) {
 		"workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}},
 	}
 	worker := map[string]*plugin_pb.ConfigValue{
-		"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}},
-		"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}},
-		"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}},
-		"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
+		"dispatch_tick_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}},
+		"checkpoint_tick_seconds":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}},
+		"refresh_interval_minutes":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}},
+		"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
+		"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
 	}
 	cfg := ParseConfig(admin, worker)
 	if cfg.Workers != 4 {
@@ -49,6 +53,9 @@ func TestParseConfigOverrides(t *testing.T) {
 	if cfg.RefreshInterval != 10*time.Minute {
 		t.Errorf("RefreshInterval=%v, want 10m", cfg.RefreshInterval)
 	}
+	if cfg.BootstrapInterval != 120*time.Minute {
+		t.Errorf("BootstrapInterval=%v, want 120m", cfg.BootstrapInterval)
+	}
 	if cfg.MaxRuntime != 120*time.Minute {
 		t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime)
 	}
@@ -56,10 +63,11 @@ func TestParseConfigOverrides(t *testing.T) {
 
 func TestParseConfigClampsZeroAndNegative(t *testing.T) {
 	worker := map[string]*plugin_pb.ConfigValue{
-		"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
-		"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
-		"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
-		"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
+		"dispatch_tick_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"checkpoint_tick_seconds":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"refresh_interval_minutes":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
+		"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
 	}
 	cfg := ParseConfig(nil, worker)
 	if cfg.DispatchTick != 1*time.Minute {
@@ -71,7 +79,23 @@ func TestParseConfigClampsZeroAndNegative(t *testing.T) {
 	if cfg.RefreshInterval != 5*time.Minute {
 		t.Errorf("zero RefreshInterval should clamp to default, got %v", cfg.RefreshInterval)
 	}
+	if cfg.BootstrapInterval != 0 {
+		t.Errorf("negative BootstrapInterval should clamp to 0, got %v", cfg.BootstrapInterval)
+	}
 	if cfg.MaxRuntime != 60*time.Minute {
 		t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime)
 	}
 }
+
+func TestParseConfigBootstrapIntervalZeroIsLegacy(t *testing.T) {
+	// Explicit zero stays zero (walk-once-per-process). Existing
+	// deployments that don't set bootstrap_interval_minutes get the
+	// legacy behavior unchanged.
+	worker := map[string]*plugin_pb.ConfigValue{
+		"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+	}
+	cfg := ParseConfig(nil, worker)
+	if cfg.BootstrapInterval != 0 {
+		t.Errorf("zero BootstrapInterval must stay zero, got %v", cfg.BootstrapInterval)
+	}
+}

weed/worker/tasks/s3_lifecycle/handler.go

@@ -122,6 +122,14 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
 							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
 						},
+						{
+							Name:        "bootstrap_interval_minutes",
+							Label:       "Bootstrap Re-walk Interval (minutes)",
+							Description: "How often each bucket is re-walked. scan_only rules — those whose retention horizon exceeds meta-log retention — only fire from the bootstrap walk, so a non-zero value is required to enforce them on a long-running worker. 0 keeps the legacy walk-once-per-process behavior.",
+							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64,
+							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
+							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+						},
 						{
 							Name:        "max_runtime_minutes",
 							Label:       "Max Runtime (minutes)",
@@ -134,10 +142,11 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 				},
 			},
 			DefaultValues: map[string]*plugin_pb.ConfigValue{
-				"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}},
-				"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}},
-				"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}},
-				"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}},
+				"dispatch_tick_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}},
+				"checkpoint_tick_seconds":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}},
+				"refresh_interval_minutes":   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}},
+				"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultBootstrapIntervalMinutes}},
+				"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}},
 			},
 		},
 		AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
@@ -240,17 +249,18 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ
 	rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn)
 
 	sched := &scheduler.Scheduler{
-		BucketsPath:     bucketsPath,
-		Engine:          engine.New(),
-		Persister:       &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
-		Client:          lifecycleRPCAdapter{c: rpc},
-		FilerClient:     filerClient,
-		ClientID:        util.RandomInt32(),
-		ClientName:      "worker-s3-lifecycle",
-		Workers:         cfg.Workers,
-		DispatchTick:    cfg.DispatchTick,
-		CheckpointTick:  cfg.CheckpointTick,
-		RefreshInterval: cfg.RefreshInterval,
+		BucketsPath:       bucketsPath,
+		Engine:            engine.New(),
+		Persister:         &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
+		Client:            lifecycleRPCAdapter{c: rpc},
+		FilerClient:       filerClient,
+		ClientID:          util.RandomInt32(),
+		ClientName:        "worker-s3-lifecycle",
+		Workers:           cfg.Workers,
+		DispatchTick:      cfg.DispatchTick,
+		CheckpointTick:    cfg.CheckpointTick,
+		RefreshInterval:   cfg.RefreshInterval,
+		BootstrapInterval: cfg.BootstrapInterval,
 	}
 	if err := sched.Run(runCtx); err != nil {
 		glog.Warningf("s3 lifecycle execute: %v", err)