feat(s3/lifecycle): plumb WalkerInterval through worker admin config (#9485)

* 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.

* feat(s3/lifecycle): plumb WalkerInterval through worker admin config

#9484 added cfg.WalkerInterval to dailyrun.Config but left the worker
side wired to zero — operators couldn't actually use the throttle
without recompiling. Add the admin-schema knob:

- New constant WalkerIntervalMinutesAdminKey = "walker_interval_minutes"
  follows the MetaLogRetentionDaysAdminKey pattern (Int64, minutes
  unit, 0 = unbounded / fire every pass).
- New Config.WalkerInterval populated in ParseConfig from
  adminValues; negative / zero stay at zero so the prior "fire every
  pass" semantics keep the in-repo integration tests and the s3tests
  sub-minute driver working unchanged.
- handler.go: admin form field with operator-facing label and
  description, default in DefaultValues, value forwarded to
  dailyrun.Run via cfg.WalkerInterval.

Tests cover the default-zero, positive, and negative cases — same
shape as the MetaLogRetention tests so the parsing contract stays
consistent.

Stacked on #9484; rebase after that lands.

weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go

@@ -30,6 +30,18 @@ const (
 	// recovery branch on the next run. 0 = unbounded (current behavior,
 	// falls back to maxTTL in runShard so PromotedHash stays empty).
 	MetaLogRetentionDaysAdminKey = "meta_log_retention_days"
+	// WalkerIntervalMinutesAdminKey throttles the per-shard steady-state
+	// and empty-replay walker fires. dailyrun.runShard checks the time
+	// since the persisted Cursor.LastWalkedNs and skips the walk when
+	// less than this interval has elapsed; cold-start and recovery walker
+	// fires (RecoveryView) stay unconditional. 0 means "fire on every
+	// run" (the prior behavior — appropriate when the worker is driven
+	// at the operator's intended walk cadence, e.g. once per hour).
+	// Production deployments running the worker at multi-second cadence
+	// (CI ticks, sub-minute admin schedules) should set this to roughly
+	// the per-shard walk budget — typically 60 (1h) for small clusters,
+	// 360+ (6h+) for large ones.
+	WalkerIntervalMinutesAdminKey = "walker_interval_minutes"
 
 	// MetadataKeyDeletesPerSecond is the per-worker share value the
 	// admin writes into ClusterContext.Metadata at ExecuteJob time.

weed/worker/tasks/s3_lifecycle/config.go

@@ -19,6 +19,11 @@ type Config struct {
 	Workers          int
 	MaxRuntime       time.Duration
 	MetaLogRetention time.Duration
+	// WalkerInterval is the minimum time between steady-state walker
+	// fires per shard. 0 means "fire on every run", preserving prior
+	// behavior; positive values gate the walker via Cursor.LastWalkedNs
+	// inside dailyrun.runShard.
+	WalkerInterval time.Duration
 }
 
 func ParseConfig(adminValues map[string]*plugin_pb.ConfigValue, workerValues map[string]*plugin_pb.ConfigValue) Config {
@@ -36,6 +41,13 @@ func ParseConfig(adminValues map[string]*plugin_pb.ConfigValue, workerValues map
 	if days := readInt64(adminValues, MetaLogRetentionDaysAdminKey, 0); days > 0 {
 		cfg.MetaLogRetention = time.Duration(days*24) * time.Hour
 	}
+	// Walker throttle. Negative / zero stay zero so dailyrun.runShard
+	// keeps the prior "fire every pass" semantics — important for in-
+	// repo integration tests and s3tests's sub-minute driver. Positive
+	// values throttle the steady-state walker per shard.
+	if mins := readInt64(adminValues, WalkerIntervalMinutesAdminKey, 0); mins > 0 {
+		cfg.WalkerInterval = time.Duration(mins) * time.Minute
+	}
 	return cfg
 }
 

weed/worker/tasks/s3_lifecycle/config_test.go

@@ -67,3 +67,37 @@ func TestParseConfigMetaLogRetentionNegativeStaysZero(t *testing.T) {
 		t.Errorf("negative MetaLogRetention should stay 0, got %v", cfg.MetaLogRetention)
 	}
 }
+
+func TestParseConfigWalkerIntervalDefaultsToZero(t *testing.T) {
+	// Unset key keeps WalkerInterval at 0 so dailyrun.runShard fires the
+	// walker every pass (the pre-throttle behavior the s3tests fast
+	// driver and the in-repo integration tests rely on).
+	cfg := ParseConfig(nil, nil)
+	if cfg.WalkerInterval != 0 {
+		t.Errorf("WalkerInterval default=%v, want 0", cfg.WalkerInterval)
+	}
+}
+
+func TestParseConfigWalkerIntervalMinutesConvertsToDuration(t *testing.T) {
+	admin := map[string]*plugin_pb.ConfigValue{
+		WalkerIntervalMinutesAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 90}},
+	}
+	cfg := ParseConfig(admin, nil)
+	if want := 90 * time.Minute; cfg.WalkerInterval != want {
+		t.Errorf("WalkerInterval=%v, want %v", cfg.WalkerInterval, want)
+	}
+}
+
+func TestParseConfigWalkerIntervalNegativeStaysZero(t *testing.T) {
+	// Negative declarations stay at 0 so the worker keeps "fire every
+	// pass" rather than treating the negative as past-due (which would
+	// fire every pass anyway — but via a less obvious code path that
+	// future readers would have to trace).
+	admin := map[string]*plugin_pb.ConfigValue{
+		WalkerIntervalMinutesAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -10}},
+	}
+	cfg := ParseConfig(admin, nil)
+	if cfg.WalkerInterval != 0 {
+		t.Errorf("negative WalkerInterval should stay 0, got %v", cfg.WalkerInterval)
+	}
+}

weed/worker/tasks/s3_lifecycle/handler.go

@@ -99,6 +99,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: 0}},
 						},
+						{
+							Name:        WalkerIntervalMinutesAdminKey,
+							Label:       "Walker Interval (minutes)",
+							Description: "Minimum time between steady-state walker fires per shard. Cold-start and rule-change recovery walks ignore this — they run unconditionally. 0 = fire on every run (use when the worker is scheduled at the desired walk cadence, e.g. hourly). Set to a positive value when the worker runs at a tighter cadence than the desired walk frequency, to avoid hammering filer with a full subtree scan per run.",
+							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}},
+						},
 					},
 				},
 			},
@@ -106,6 +114,7 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 				ClusterDeletesPerSecondAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
 				ClusterDeletesBurstAdminKey:     {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
 				MetaLogRetentionDaysAdminKey:    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+				WalkerIntervalMinutesAdminKey:   {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
 			},
 		},
 		WorkerConfigForm: &plugin_pb.ConfigForm{
@@ -295,6 +304,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe
 		Limiter:         limiter,
 		RetentionWindow: cfg.MetaLogRetention,
 		Walker:          walker,
+		WalkerInterval:  cfg.WalkerInterval,
 		ClientName:      "worker-s3-lifecycle-daily",
 	})
 	if runErr != nil {