From bbc075b3533240d20287191639f74594b4961b60 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 13 May 2026 14:09:31 -0700 Subject: [PATCH] feat(s3/lifecycle): plumb WalkerInterval through worker admin config (#9485) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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. --- .../tasks/s3_lifecycle/cluster_rate_limit.go | 12 +++++++ weed/worker/tasks/s3_lifecycle/config.go | 12 +++++++ weed/worker/tasks/s3_lifecycle/config_test.go | 34 +++++++++++++++++++ weed/worker/tasks/s3_lifecycle/handler.go | 10 ++++++ 4 files changed, 68 insertions(+) diff --git a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go index d927f2b37..34866a38e 100644 --- a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go +++ b/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. diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go index 6bf38dd11..a041194ee 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/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 } diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go index 287c9fb74..ae6bee00f 100644 --- a/weed/worker/tasks/s3_lifecycle/config_test.go +++ b/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) + } +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 947036036..3b4aad99f 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/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 {