From 0dde6a8c844caef4d60339fcc66bc8f328427e6c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 13 May 2026 19:29:06 -0700 Subject: [PATCH] refactor(s3/lifecycle): drop Per-Run Time Limit knob; use scheduler's Execution Timeout (#9494) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(s3/lifecycle): drop Per-Run Time Limit knob; use scheduler's Execution Timeout "Per-Run Time Limit (minutes)" duplicated the admin scheduler's "Execution Timeout (s)" — both are wall-clock caps on the same Execute call, stacked via context.WithTimeout. Whichever was shorter won. Under defaults the scheduler's 90s timeout always clobbered the worker's 60-min cap, so the "Per-Run Time Limit" knob was effectively dead unless an operator also raised Execution Timeout, and operators had to keep two values in agreement. Remove the worker-side knob and declare a sane scheduler default on the handler descriptor: - WorkerConfigForm: nil (was: one section with one field) - Config.MaxRuntime removed; ParseConfig drops max_runtime_minutes - Handler no longer wraps ctx in context.WithTimeout(MaxRuntime); runCtx is just the ctx the scheduler passes - AdminRuntimeDefaults.ExecutionTimeoutSeconds = 3600 (1h) and JobTypeMaxRuntimeSeconds = 3600 — the scheduler's global 90s default would otherwise kill every real run Tests: - TestParseConfigDefaults loses the MaxRuntime check; new TestParseConfigIgnoresWorkerValues documents the contract - TestDescriptor_WorkerConfigFormIsAbsent pins that the form is gone so a future re-add forces a conscious revisit - TestDescriptor_AdminRuntimeDefaultsBoundExecutionTimeout pins the 1h default with a comment about the 90s scheduler floor * fix(s3/lifecycle): no per-pass timeout by default Lifecycle is a scheduled batch — its natural duration is "as long as today's events take." The 1h default ExecutionTimeoutSeconds from the previous commit was still a footgun: too low truncates legitimate large-bucket passes; too high makes the value meaningless. Set both ExecutionTimeoutSeconds and JobTypeMaxRuntimeSeconds to math.MaxInt32 (~68 years) to say "no timeout in practice" in a code-review-readable way. Operators who genuinely want a wall-clock cap can set one in the admin UI; the scheduler's context.WithTimeout machinery is unchanged (we just hand it an effectively-infinite duration). Note: the scheduler floors ExecutionTimeout at 90s (defaultScheduledExecutionTimeout in weed/admin/plugin/plugin_scheduler.go), so 0 doesn't mean "unlimited" — it clamps back to 90s. A literal math.MaxInt32 is the way to express the intent without touching the shared scheduler code. Test updated to pin math.MaxInt32 and document the rationale so a future tighter cap fails the test and forces conscious revisit. --- weed/worker/tasks/s3_lifecycle/config.go | 10 +-- weed/worker/tasks/s3_lifecycle/config_test.go | 24 +++---- weed/worker/tasks/s3_lifecycle/handler.go | 63 ++++++++++--------- .../worker/tasks/s3_lifecycle/handler_test.go | 57 +++++++++-------- 4 files changed, 71 insertions(+), 83 deletions(-) diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go index a041194ee..6b21c2bcb 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -11,13 +11,10 @@ const ( // In-process fan-out across the 16 shards. shardPipelineGoroutines = 1 - - defaultMaxRuntimeMinutes = int64(60) ) 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 @@ -28,12 +25,9 @@ type Config struct { func ParseConfig(adminValues map[string]*plugin_pb.ConfigValue, workerValues map[string]*plugin_pb.ConfigValue) Config { cfg := Config{ - Workers: shardPipelineGoroutines, - MaxRuntime: time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute, - } - if cfg.MaxRuntime <= 0 { - cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute + Workers: shardPipelineGoroutines, } + _ = workerValues // worker-side config form is currently empty; reserved for future per-worker tuning. // Operator-declared meta-log retention. Negative or zero values stay // zero so runShard falls back to maxTTL (PromotedHash dormant). // Convert days->hours in int64 space before lifting to time.Duration diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go index ae6bee00f..4b774f345 100644 --- a/weed/worker/tasks/s3_lifecycle/config_test.go +++ b/weed/worker/tasks/s3_lifecycle/config_test.go @@ -12,28 +12,20 @@ func TestParseConfigDefaults(t *testing.T) { if cfg.Workers != shardPipelineGoroutines { t.Errorf("Workers default=%d, want %d", cfg.Workers, shardPipelineGoroutines) } - if cfg.MaxRuntime != 60*time.Minute { - t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime) - } } -func TestParseConfigOverrideMaxRuntime(t *testing.T) { +func TestParseConfigIgnoresWorkerValues(t *testing.T) { + // Worker-side config form is currently empty. The old + // max_runtime_minutes knob duplicated the admin scheduler's + // Execution Timeout and was removed; anything in workerValues is + // now silently ignored. The single source of truth for the + // per-Execute wall-clock cap is AdminRuntimeDefaults.ExecutionTimeoutSeconds. worker := map[string]*plugin_pb.ConfigValue{ "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, } cfg := ParseConfig(nil, worker) - if cfg.MaxRuntime != 120*time.Minute { - t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime) - } -} - -func TestParseConfigNegativeMaxRuntimeClampsToDefault(t *testing.T) { - worker := map[string]*plugin_pb.ConfigValue{ - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, - } - cfg := ParseConfig(nil, worker) - if cfg.MaxRuntime != 60*time.Minute { - t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime) + if cfg.Workers != shardPipelineGoroutines { + t.Errorf("Workers default=%d, want %d", cfg.Workers, shardPipelineGoroutines) } } diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index caaf125b7..57a6f751a 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -3,6 +3,7 @@ package s3_lifecycle import ( "context" "fmt" + "math" "strconv" "time" @@ -117,31 +118,12 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { WalkerIntervalMinutesAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, }, }, - WorkerConfigForm: &plugin_pb.ConfigForm{ - FormId: "s3-lifecycle-worker", - Title: "S3 Lifecycle Worker", - Description: "Operational tuning for the daily lifecycle run.", - Sections: []*plugin_pb.ConfigSection{ - { - SectionId: "cadence", - Title: "Cadence", - Description: "Per-run wall-clock budget.", - Fields: []*plugin_pb.ConfigField{ - { - Name: "max_runtime_minutes", - Label: "Per-Run Time Limit (minutes)", - Description: "Wall-clock cap on each run. Each daily run processes one day of meta-log events plus the walker pass over the current rule set; the cursor persists across runs.", - 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: 1}}, - }, - }, - }, - }, - DefaultValues: map[string]*plugin_pb.ConfigValue{ - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}}, - }, - }, + // WorkerConfigForm intentionally absent — the prior + // "Per-Run Time Limit (minutes)" knob duplicated the admin + // scheduler's Execution Timeout (both wall-clock caps on the + // same Execute call), and operators had to keep the two values + // in agreement. Removed in favor of a single source of truth: + // AdminRuntimeDefaults.ExecutionTimeoutSeconds below. AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ // On by default: S3 lifecycle is a standard bucket feature // (PutBucketLifecycleConfiguration is part of the S3 API), @@ -155,6 +137,21 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { DetectionIntervalMinutes: 24 * 60, // daily DetectionTimeoutSeconds: 60, MaxJobsPerDetection: 1, + // Effectively no per-pass wall-clock cap. Lifecycle is a + // scheduled batch — its natural duration is "as long as it + // takes to process today's events." The scheduler's global + // 90s default would kill every real run, and a numeric + // cap operators have to estimate (1h? 8h?) is a recurring + // footgun: too low truncates a legitimate large-bucket + // pass; too high makes the value meaningless. + // + // Use math.MaxInt32 seconds (~68 years) for both knobs to + // say "no timeout in practice" in code-review-readable form. + // Operators who genuinely want a cap can set one in the + // admin UI; the underlying context.WithTimeout machinery + // is unchanged. + ExecutionTimeoutSeconds: math.MaxInt32, + JobTypeMaxRuntimeSeconds: math.MaxInt32, }, } } @@ -218,13 +215,19 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ return fmt.Errorf("execute: missing filer_grpc_address in job parameters") } - runCtx, cancel := context.WithTimeout(ctx, cfg.MaxRuntime) - defer cancel() + // Run lifetime is bounded by the scheduler's Execution Timeout + // (admin UI). The handler used to wrap ctx in another + // context.WithTimeout(cfg.MaxRuntime), but that doubled the + // concept — both knobs were wall-clock caps on the same Execute + // call, and the smaller one always won (typically the 90s + // scheduler default would clobber a 60-min worker setting). + // Single source of truth is now AdminRuntimeDefaults.ExecutionTimeoutSeconds. + runCtx := ctx _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ JobId: request.Job.JobId, JobType: jobType, State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting", - Message: fmt.Sprintf("scheduler workers=%d s3=%v runtime=%s", cfg.Workers, s3Endpoints, cfg.MaxRuntime), + Message: fmt.Sprintf("scheduler workers=%d s3=%v", cfg.Workers, s3Endpoints), }) dialCtx, dialCancel := context.WithTimeout(runCtx, 30*time.Second) @@ -297,8 +300,8 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ JobId: request.Job.JobId, JobType: jobType, State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting", - Message: fmt.Sprintf("daily_replay shards=%d workers=%d runtime=%s rate=%s buckets=%d walker=on", - len(shards), cfg.Workers, cfg.MaxRuntime, limiterDesc, len(buckets)), + Message: fmt.Sprintf("daily_replay shards=%d workers=%d rate=%s buckets=%d walker=on", + len(shards), cfg.Workers, limiterDesc, len(buckets)), }) runErr := dailyrun.Run(ctx, dailyrun.Config{ diff --git a/weed/worker/tasks/s3_lifecycle/handler_test.go b/weed/worker/tasks/s3_lifecycle/handler_test.go index 5ed1beca4..af715f21b 100644 --- a/weed/worker/tasks/s3_lifecycle/handler_test.go +++ b/weed/worker/tasks/s3_lifecycle/handler_test.go @@ -3,6 +3,7 @@ package s3_lifecycle import ( "context" "errors" + "math" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -301,38 +302,36 @@ func TestDescriptor_AdminConfigFormHasNoWorkersField(t *testing.T) { assert.False(t, hasDefault, "DefaultValues must NOT include 'workers' (form field removed)") } -func TestDescriptor_WorkerConfigFormCadenceDefaultsMatchParseConfig(t *testing.T) { - // Every default the parser reads must be exposed in the descriptor's - // DefaultValues; otherwise the admin UI would seed the form with a - // blank or zero value and the worker would silently clamp to the - // hardcoded fallback. Drift between the two is the bug this test - // catches. +func TestDescriptor_WorkerConfigFormIsAbsent(t *testing.T) { + // "Per-Run Time Limit (minutes)" was the only worker-side knob and + // duplicated the admin scheduler's Execution Timeout (both are + // wall-clock caps on the same Execute call). Removed in favor of + // AdminRuntimeDefaults.ExecutionTimeoutSeconds — single source of + // truth. A WorkerConfigForm with no fields would render as an + // empty section in the admin UI; drop the form entirely. h := NewHandler(nil) d := h.Descriptor() - require.NotNil(t, d.WorkerConfigForm) - assert.Equal(t, "s3-lifecycle-worker", d.WorkerConfigForm.FormId) + assert.Nil(t, d.WorkerConfigForm, + "WorkerConfigForm should be nil now that max_runtime_minutes is gone; if you re-add a worker-side knob, restore the form and pin it here") +} - wantDefaults := map[string]int64{ - "max_runtime_minutes": defaultMaxRuntimeMinutes, - } - for name, want := range wantDefaults { - t.Run(name, func(t *testing.T) { - dv, ok := d.WorkerConfigForm.DefaultValues[name] - require.True(t, ok, "WorkerConfigForm.DefaultValues missing %q", name) - assert.Equal(t, want, dv.GetInt64Value(), "default mismatch for %q", name) - }) - } - declared := map[string]plugin_pb.ConfigFieldType{} - for _, sec := range d.WorkerConfigForm.Sections { - for _, f := range sec.Fields { - declared[f.Name] = f.FieldType - } - } - for name := range wantDefaults { - ft, ok := declared[name] - assert.True(t, ok, "WorkerConfigForm has no field named %q", name) - assert.Equal(t, plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, ft, "field %q must be INT64 to match readInt64", name) - } +func TestDescriptor_AdminRuntimeDefaultsHaveNoTimeoutInPractice(t *testing.T) { + // Lifecycle is a scheduled batch whose natural duration is "as + // long as today's events take." The scheduler's global 90s default + // would kill every real run, and a numeric cap operators have to + // estimate is a footgun (too low truncates a legitimate + // large-bucket pass; too high makes the value meaningless). Declare + // math.MaxInt32 seconds for both knobs to say "no timeout in + // practice" in a code-review-readable way. A future change that + // tightens the cap should fail this test so the choice is + // re-examined consciously. + h := NewHandler(nil) + d := h.Descriptor() + require.NotNil(t, d.AdminRuntimeDefaults) + assert.Equal(t, int32(math.MaxInt32), d.AdminRuntimeDefaults.ExecutionTimeoutSeconds, + "ExecutionTimeoutSeconds should be effectively unlimited; the scheduler's 90s default would otherwise clobber the worker mid-run") + assert.Equal(t, int32(math.MaxInt32), d.AdminRuntimeDefaults.JobTypeMaxRuntimeSeconds, + "JobTypeMaxRuntimeSeconds is the per-pass budget — keep aligned with ExecutionTimeoutSeconds") } func TestDescriptor_AdminRuntimeDefaultsDailyCadence(t *testing.T) {