From ce5768fab1d8a2407d072eb709dffa0615ba7094 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 12 May 2026 18:26:52 -0700 Subject: [PATCH] feat(s3/lifecycle): operator-declared meta-log retention activates PromotedHash (#9473) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): operator-declared meta-log retention activates PromotedHash dailyrun.Config.RetentionWindow has been wired since Phase 4b but the handler never supplied a value, so runShard always fell back to maxTTL and engine.PromotedHash hashed nothing. The partition-flip recovery trigger was dormant by design "until the handler plumbs the real meta-log retention here." This PR plumbs it via a new admin form field: Meta-Log Retention (days) — 0 = unbounded (current behavior). When set, ParseConfig converts days to a time.Duration on cfg.MetaLogRetention. The handler passes it as dailyrun.Config.RetentionWindow, which runShard then feeds to engine.PromotedHash. Rules whose TTL exceeds the declared window land in the walk partition; the next time an operator shrinks retention so a previously replay-eligible rule slips past it, PromotedHash mismatches → recovery branch fires → walker re-evaluates the rule across the whole filer tree. 0 stays the default, so existing deployments see no behavior change. * chore(s3/lifecycle): rephrase days->duration conversion gemini-code-assist flagged the original form as a compile error, which it wasn't (time.Duration is a named int64 and supports * with other time.Durations — the test suite verified the value was correct). The suggested form is more idiomatic regardless: days*24 happens in int64 space before the lift to time.Duration, so the unit is unambiguous. --- .../tasks/s3_lifecycle/cluster_rate_limit.go | 8 +++++ weed/worker/tasks/s3_lifecycle/config.go | 14 ++++++-- weed/worker/tasks/s3_lifecycle/config_test.go | 31 ++++++++++++++++++ weed/worker/tasks/s3_lifecycle/handler.go | 32 ++++++++++++------- 4 files changed, 71 insertions(+), 14 deletions(-) diff --git a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go index 1e92dd6df..d927f2b37 100644 --- a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go +++ b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go @@ -22,6 +22,14 @@ const ( // ClusterDeletesBurstAdminKey holds the token-bucket burst. 0 means // "2 × rps" (computed by the admin allocator). ClusterDeletesBurstAdminKey = "cluster_deletes_burst" + // MetaLogRetentionDaysAdminKey holds the operator's declaration of + // how far back the filer's meta-log subscription can reliably reach. + // Rules whose effective TTL exceeds this window can't be serviced by + // replay alone and get partitioned into engine.PromotedHash's walk + // set; a partition flip (operator shrinks retention) then trips the + // 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" // 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 5831e3581..6bf38dd11 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -16,11 +16,12 @@ const ( ) type Config struct { - Workers int - MaxRuntime time.Duration + Workers int + MaxRuntime time.Duration + MetaLogRetention time.Duration } -func ParseConfig(_ map[string]*plugin_pb.ConfigValue, workerValues map[string]*plugin_pb.ConfigValue) Config { +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, @@ -28,6 +29,13 @@ func ParseConfig(_ map[string]*plugin_pb.ConfigValue, workerValues map[string]*p if cfg.MaxRuntime <= 0 { cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute } + // 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 + // so the unit is unambiguous. + if days := readInt64(adminValues, MetaLogRetentionDaysAdminKey, 0); days > 0 { + cfg.MetaLogRetention = time.Duration(days*24) * time.Hour + } return cfg } diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go index 4e5d919aa..287c9fb74 100644 --- a/weed/worker/tasks/s3_lifecycle/config_test.go +++ b/weed/worker/tasks/s3_lifecycle/config_test.go @@ -36,3 +36,34 @@ func TestParseConfigNegativeMaxRuntimeClampsToDefault(t *testing.T) { t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime) } } + +func TestParseConfigMetaLogRetentionDefaultsToZero(t *testing.T) { + // Unset key keeps MetaLogRetention at 0, which runShard treats as + // "no retention info supplied" and falls back to maxTTL. + cfg := ParseConfig(nil, nil) + if cfg.MetaLogRetention != 0 { + t.Errorf("MetaLogRetention default=%v, want 0", cfg.MetaLogRetention) + } +} + +func TestParseConfigMetaLogRetentionDaysConvertsToDuration(t *testing.T) { + admin := map[string]*plugin_pb.ConfigValue{ + MetaLogRetentionDaysAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 7}}, + } + cfg := ParseConfig(admin, nil) + if want := 7 * 24 * time.Hour; cfg.MetaLogRetention != want { + t.Errorf("MetaLogRetention=%v, want %v (7 days)", cfg.MetaLogRetention, want) + } +} + +func TestParseConfigMetaLogRetentionNegativeStaysZero(t *testing.T) { + // A negative declaration is nonsense; stay at 0 so runShard's + // fallback applies rather than producing a negative window. + admin := map[string]*plugin_pb.ConfigValue{ + MetaLogRetentionDaysAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -3}}, + } + cfg := ParseConfig(admin, nil) + if cfg.MetaLogRetention != 0 { + t.Errorf("negative MetaLogRetention should stay 0, got %v", cfg.MetaLogRetention) + } +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index c29de8c3f..5fe2fa0d1 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -91,12 +91,21 @@ 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: MetaLogRetentionDaysAdminKey, + Label: "Meta-Log Retention (days)", + Description: "How far back the filer's meta-log subscription can reach. Rules whose TTL exceeds this run via the walker; shrinking this value will trigger a one-time recovery walk on the next run for any rule that's now too old to replay. 0 = unbounded (no partition; every rule serviced by replay).", + 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}}, + }, }, }, }, DefaultValues: map[string]*plugin_pb.ConfigValue{ ClusterDeletesPerSecondAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, ClusterDeletesBurstAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + MetaLogRetentionDaysAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, }, }, WorkerConfigForm: &plugin_pb.ConfigForm{ @@ -273,17 +282,18 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe }) runErr := dailyrun.Run(ctx, dailyrun.Config{ - Shards: shards, - BucketsPath: bucketsPath, - Engine: eng, - FilerClient: filerClient, - Client: client, - Persister: &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, - Lister: dispatcher.NewFilerSiblingLister(filerClient, bucketsPath), - Workers: cfg.Workers, - Limiter: limiter, - Walker: walker, - ClientName: "worker-s3-lifecycle-daily", + Shards: shards, + BucketsPath: bucketsPath, + Engine: eng, + FilerClient: filerClient, + Client: client, + Persister: &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, + Lister: dispatcher.NewFilerSiblingLister(filerClient, bucketsPath), + Workers: cfg.Workers, + Limiter: limiter, + RetentionWindow: cfg.MetaLogRetention, + Walker: walker, + ClientName: "worker-s3-lifecycle-daily", }) if runErr != nil { glog.Warningf("daily_replay: %v", runErr)