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)