feat(s3/lifecycle): operator-declared meta-log retention activates PromotedHash (#9473)

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

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.

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
 }
 

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)
+	}
+}

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)