feat(worker/s3_lifecycle): plugin handler with admin UI config (#9362)

* feat(s3/lifecycle): scheduler — N pipelines over an even shard split

Scheduler.Run spawns Workers Pipeline goroutines plus one engine-refresh
ticker. Each worker owns a contiguous AssignShards(idx, total) slice of
[0, ShardCount) and runs Pipeline.Run with EventBudget bounding each
iteration; brief RetryBackoff between iterations avoids hot-loop on
errors. The refresh ticker rebuilds the engine snapshot from the filer's
bucket configs every RefreshInterval.

LoadCompileInputs / IsBucketVersioned / AllActivePriorStates are
exported from a configload.go sibling so the shell command can move to
this shared implementation in a follow-up.

* refactor(shell): reuse scheduler.LoadCompileInputs in run-shard

Drop the local copies of loadLifecycleCompileInputs / isBucketVersioned
/ allActivePriorStates / lifecycleParseError that the new
scheduler package now exports. Same behavior, one source of truth.

* feat(worker/s3_lifecycle): plugin handler with admin UI config

Registers a JobHandler for s3_lifecycle via pluginworker.RegisterHandler.
Admin pulls the descriptor over the worker plugin gRPC and renders the
AdminConfigForm + WorkerConfigForm in the existing UI:

  Admin form (cluster shape):
    - workers (1..16, default 1)
    - s3_grpc_endpoints (comma list)

  Worker form (operational tuning):
    - dispatch_tick_ms (default 5000)
    - checkpoint_tick_ms (default 30000)
    - refresh_interval_ms (default 300000)
    - event_budget (default 0 = unbounded)

Detect emits a single proposal whenever S3 endpoints + filer addresses
are configured. MaxExecutionConcurrency=1 so admin only ever runs one
lifecycle daemon per worker; a fresh proposal next cycle restarts it
if the prior Execute exits.

Execute dials the configured S3 endpoint + filer, builds a
scheduler.Scheduler with the parsed config, and runs it until
ctx cancellation. Reuses the existing scheduler / dispatcher /
reader / engine packages — the handler is the thin glue that
parses descriptor values and wires the long-running daemon.

* proto(plugin): add s3_grpc_addresses to ClusterContext

So workers can dial s3 servers discovered by the master rather than a
hand-typed list in the admin form.

* feat(admin): populate ClusterContext.s3_grpc_addresses from master

ListClusterNodes(S3Type) returns the live S3 servers; the plugin
scheduler now hands these to job handlers alongside filer/volume
addresses.

* feat(worker/s3_lifecycle): discover s3 endpoints from cluster context

Drop the s3_grpc_endpoints admin form field and read the master-supplied
ClusterContext.S3GrpcAddresses instead. Operators no longer maintain a
hand-typed list, and a stale entry self-heals when the master's view
updates.

* feat(worker/s3_lifecycle): time-based runtime cap, friendlier cadence units

- dispatch_tick_minutes (was *_ms): minutes is the natural granularity
  for a daily batch; default 1 minute.
- checkpoint_tick_seconds: seconds for the durable cursor write; default
  30 seconds.
- refresh_interval_minutes: minutes for the engine snapshot rebuild.
- max_runtime_minutes replaces event_budget. Each daily run is bounded
  by wall clock — typical run wraps in well under an hour because the
  cursor persists and the meta-log streams fast. Default 60 minutes.
- AdminRuntimeDefaults.DetectionIntervalSeconds = 86400 so the admin
  schedules one job per day.

weed/admin/dash/plugin_api.go

@@ -14,7 +14,9 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/seaweedfs/seaweedfs/weed/admin/plugin"
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
@@ -738,6 +740,9 @@ func (s *AdminServer) parseOrBuildClusterContext(raw json.RawMessage) (*plugin_p
 	if len(contextMessage.VolumeGrpcAddresses) == 0 {
 		contextMessage.VolumeGrpcAddresses = append(contextMessage.VolumeGrpcAddresses, fallback.VolumeGrpcAddresses...)
 	}
+	if len(contextMessage.S3GrpcAddresses) == 0 {
+		contextMessage.S3GrpcAddresses = append(contextMessage.S3GrpcAddresses, fallback.S3GrpcAddresses...)
+	}
 	if contextMessage.Metadata == nil {
 		contextMessage.Metadata = map[string]string{}
 	}
@@ -751,6 +756,7 @@ func (s *AdminServer) buildDefaultPluginClusterContext() *plugin_pb.ClusterConte
 		MasterGrpcAddresses: make([]string, 0),
 		FilerGrpcAddresses:  make([]string, 0),
 		VolumeGrpcAddresses: make([]string, 0),
+		S3GrpcAddresses:     make([]string, 0),
 		Metadata: map[string]string{
 			"source": "admin",
 		},
@@ -794,9 +800,34 @@ func (s *AdminServer) buildDefaultPluginClusterContext() *plugin_pb.ClusterConte
 		glog.V(1).Infof("failed to build default plugin volume context: %v", err)
 	}
 
+	s3Seen := map[string]struct{}{}
+	if err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
+		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+			ClientType: cluster.S3Type,
+		})
+		if err != nil {
+			return err
+		}
+		for _, node := range resp.GetClusterNodes() {
+			address := strings.TrimSpace(node.GetAddress())
+			if address == "" {
+				continue
+			}
+			if _, exists := s3Seen[address]; exists {
+				continue
+			}
+			s3Seen[address] = struct{}{}
+			clusterContext.S3GrpcAddresses = append(clusterContext.S3GrpcAddresses, address)
+		}
+		return nil
+	}); err != nil {
+		glog.V(1).Infof("failed to list cluster S3 nodes: %v", err)
+	}
+
 	sort.Strings(clusterContext.MasterGrpcAddresses)
 	sort.Strings(clusterContext.FilerGrpcAddresses)
 	sort.Strings(clusterContext.VolumeGrpcAddresses)
+	sort.Strings(clusterContext.S3GrpcAddresses)
 
 	return clusterContext
 }

weed/pb/plugin.proto

@@ -351,6 +351,7 @@ message ClusterContext {
   repeated string filer_grpc_addresses = 2;
   repeated string volume_grpc_addresses = 3;
   map<string, string> metadata = 4;
+  repeated string s3_grpc_addresses = 5;
 }
 
 message ActivityEvent {

weed/pb/plugin_pb/plugin.pb.go

@@ -3567,6 +3567,7 @@ type ClusterContext struct {
 	FilerGrpcAddresses  []string               `protobuf:"bytes,2,rep,name=filer_grpc_addresses,json=filerGrpcAddresses,proto3" json:"filer_grpc_addresses,omitempty"`
 	VolumeGrpcAddresses []string               `protobuf:"bytes,3,rep,name=volume_grpc_addresses,json=volumeGrpcAddresses,proto3" json:"volume_grpc_addresses,omitempty"`
 	Metadata            map[string]string      `protobuf:"bytes,4,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
+	S3GrpcAddresses     []string               `protobuf:"bytes,5,rep,name=s3_grpc_addresses,json=s3GrpcAddresses,proto3" json:"s3_grpc_addresses,omitempty"`
 	unknownFields       protoimpl.UnknownFields
 	sizeCache           protoimpl.SizeCache
 }
@@ -3629,6 +3630,13 @@ func (x *ClusterContext) GetMetadata() map[string]string {
 	return nil
 }
 
+func (x *ClusterContext) GetS3GrpcAddresses() []string {
+	if x != nil {
+		return x.S3GrpcAddresses
+	}
+	return nil
+}
+
 type ActivityEvent struct {
 	state         protoimpl.MessageState  `protogen:"open.v1"`
 	Source        ActivitySource          `protobuf:"varint,1,opt,name=source,proto3,enum=plugin.ActivitySource" json:"source,omitempty"`
@@ -4265,12 +4273,13 @@ const file_plugin_proto_rawDesc = "" +
 	"\asummary\x18\x02 \x01(\tR\asummary\x1aT\n" +
 	"\x11OutputValuesEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12)\n" +
-	"\x05value\x18\x02 \x01(\v2\x13.plugin.ConfigValueR\x05value:\x028\x01\"\xa9\x02\n" +
+	"\x05value\x18\x02 \x01(\v2\x13.plugin.ConfigValueR\x05value:\x028\x01\"\xd5\x02\n" +
 	"\x0eClusterContext\x122\n" +
 	"\x15master_grpc_addresses\x18\x01 \x03(\tR\x13masterGrpcAddresses\x120\n" +
 	"\x14filer_grpc_addresses\x18\x02 \x03(\tR\x12filerGrpcAddresses\x122\n" +
 	"\x15volume_grpc_addresses\x18\x03 \x03(\tR\x13volumeGrpcAddresses\x12@\n" +
-	"\bmetadata\x18\x04 \x03(\v2$.plugin.ClusterContext.MetadataEntryR\bmetadata\x1a;\n" +
+	"\bmetadata\x18\x04 \x03(\v2$.plugin.ClusterContext.MetadataEntryR\bmetadata\x12*\n" +
+	"\x11s3_grpc_addresses\x18\x05 \x03(\tR\x0fs3GrpcAddresses\x1a;\n" +
 	"\rMetadataEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xb9\x02\n" +

weed/plugin/worker/handlers/handlers.go

@@ -9,5 +9,6 @@ import (
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance"     // register ec_balance handler
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding" // register erasure_coding handler
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/iceberg"        // register iceberg_maintenance handler
+	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/s3_lifecycle"   // register s3_lifecycle handler
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"         // register vacuum handler
 )

weed/s3api/s3lifecycle/scheduler/configload.go

@@ -0,0 +1,103 @@
+package scheduler
+
+import (
+	"context"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/lifecycle_xml"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+)
+
+const BucketLifecycleConfigurationXMLKey = "s3-bucket-lifecycle-configuration-xml"
+
+// ParseError is returned alongside successfully-loaded inputs so callers can
+// surface malformed bucket configs rather than silently dropping them.
+type ParseError struct {
+	Bucket string
+	Err    error
+}
+
+// LoadCompileInputs walks the buckets directory and returns one
+// engine.CompileInput per bucket that carries a non-empty lifecycle
+// configuration. Pagination loops with startFrom so clusters with more
+// than one page of buckets don't drop the tail.
+func LoadCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketsPath string) ([]engine.CompileInput, []ParseError, error) {
+	var (
+		inputs      []engine.CompileInput
+		parseErrors []ParseError
+		startFrom   string
+	)
+	const pageSize uint32 = 1024
+	for {
+		pageCount := 0
+		var lastName string
+		err := filer_pb.SeaweedList(ctx, client, bucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+			pageCount++
+			lastName = entry.Name
+			if !entry.IsDirectory {
+				return nil
+			}
+			xmlBytes, ok := entry.Extended[BucketLifecycleConfigurationXMLKey]
+			if !ok || len(xmlBytes) == 0 {
+				return nil
+			}
+			rules, err := lifecycle_xml.ParseCanonical(xmlBytes)
+			if err != nil {
+				parseErrors = append(parseErrors, ParseError{Bucket: entry.Name, Err: err})
+				return nil
+			}
+			if len(rules) == 0 {
+				return nil
+			}
+			inputs = append(inputs, engine.CompileInput{
+				Bucket:    entry.Name,
+				Rules:     rules,
+				Versioned: IsBucketVersioned(entry),
+			})
+			return nil
+		}, startFrom, false, pageSize)
+		if err != nil {
+			return nil, nil, err
+		}
+		if uint32(pageCount) < pageSize {
+			break
+		}
+		startFrom = lastName
+	}
+	return inputs, parseErrors, nil
+}
+
+// IsBucketVersioned returns true when the bucket's filer entry has the
+// versioning extended attribute set to Enabled or Suspended.
+func IsBucketVersioned(entry *filer_pb.Entry) bool {
+	v, ok := entry.Extended[s3_constants.ExtVersioningKey]
+	if !ok {
+		return false
+	}
+	s := strings.ToLower(strings.TrimSpace(string(v)))
+	return s == "enabled" || s == "suspended"
+}
+
+// AllActivePriorStates seeds every compiled action as bootstrap-complete +
+// event-driven so the scheduler dispatches whatever fires immediately.
+// Production bootstrap walks set this incrementally per bucket; the
+// scheduler runs out-of-band of that flow for now.
+func AllActivePriorStates(inputs []engine.CompileInput) map[s3lifecycle.ActionKey]engine.PriorState {
+	prior := map[s3lifecycle.ActionKey]engine.PriorState{}
+	for _, in := range inputs {
+		for _, rule := range in.Rules {
+			hash := s3lifecycle.RuleHash(rule)
+			for _, kind := range s3lifecycle.RuleActionKinds(rule) {
+				key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: hash, ActionKind: kind}
+				prior[key] = engine.PriorState{
+					BootstrapComplete: true,
+					Mode:              engine.ModeEventDriven,
+				}
+			}
+		}
+	}
+	return prior
+}

weed/s3api/s3lifecycle/scheduler/scheduler.go

@@ -0,0 +1,166 @@
+package scheduler
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
+)
+
+const (
+	defaultRefreshInterval = 5 * time.Minute
+	defaultRetryBackoff    = 5 * time.Second
+)
+
+// Scheduler runs N pipeline goroutines, one per worker, each owning a
+// contiguous shard slice of [0, ShardCount). It periodically rebuilds the
+// engine snapshot from the filer's bucket configs.
+type Scheduler struct {
+	BucketsPath string
+	Engine      *engine.Engine
+	Persister   reader.Persister
+	Client      dispatcher.LifecycleClient
+	FilerClient filer_pb.SeaweedFilerClient
+	ClientID    int32
+	ClientName  string
+
+	Workers         int
+	DispatchTick    time.Duration
+	CheckpointTick  time.Duration
+	RefreshInterval time.Duration
+	RetryBackoff    time.Duration
+}
+
+// Run blocks until ctx is canceled. Spawns Workers + 1 goroutines: one
+// engine-refresh ticker plus one Pipeline runner per worker.
+func (s *Scheduler) Run(ctx context.Context) error {
+	if s.Engine == nil || s.Persister == nil || s.Client == nil || s.FilerClient == nil {
+		return errors.New("scheduler: missing required dependency")
+	}
+	if s.BucketsPath == "" {
+		return errors.New("scheduler: BucketsPath required")
+	}
+
+	workers := s.Workers
+	if workers <= 0 {
+		workers = 1
+	}
+	refresh := s.RefreshInterval
+	if refresh <= 0 {
+		refresh = defaultRefreshInterval
+	}
+	retry := s.RetryBackoff
+	if retry <= 0 {
+		retry = defaultRetryBackoff
+	}
+
+	s.refreshEngine(ctx)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		t := time.NewTicker(refresh)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				s.refreshEngine(ctx)
+			}
+		}
+	}()
+
+	for i := 0; i < workers; i++ {
+		shards := AssignShards(i, workers)
+		if len(shards) == 0 {
+			continue
+		}
+		wg.Add(1)
+		go func(idx int, shardSet []int) {
+			defer wg.Done()
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+				p := &dispatcher.Pipeline{
+					Shards:         shardSet,
+					BucketsPath:    s.BucketsPath,
+					Engine:         s.Engine,
+					Persister:      s.Persister,
+					Client:         s.Client,
+					FilerClient:    s.FilerClient,
+					ClientID:       s.ClientID,
+					ClientName:     fmt.Sprintf("%s-w%02d", s.ClientName, idx),
+					DispatchTick:   s.DispatchTick,
+					CheckpointTick: s.CheckpointTick,
+				}
+				if err := p.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
+					glog.Warningf("lifecycle scheduler: worker=%d shards=%v: %v", idx, shardSet, err)
+				}
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(retry):
+				}
+			}
+		}(i, shards)
+	}
+
+	wg.Wait()
+	return nil
+}
+
+func (s *Scheduler) refreshEngine(ctx context.Context) {
+	inputs, parseErrors, err := LoadCompileInputs(ctx, s.FilerClient, s.BucketsPath)
+	if err != nil {
+		glog.Warningf("lifecycle scheduler: refresh engine: %v", err)
+		return
+	}
+	for _, pe := range parseErrors {
+		glog.Warningf("lifecycle scheduler: malformed config in bucket %s: %v", pe.Bucket, pe.Err)
+	}
+	s.Engine.Compile(inputs, engine.CompileOptions{PriorStates: AllActivePriorStates(inputs)})
+}
+
+// AssignShards returns the contiguous shard slice for worker idx of total.
+// Shards distribute as evenly as possible: with ShardCount=16 and total=3,
+// workers receive [0..5], [6..10], [11..15].
+func AssignShards(idx, total int) []int {
+	if total <= 0 || idx < 0 || idx >= total {
+		return nil
+	}
+	shardCount := s3lifecycle.ShardCount
+	base := shardCount / total
+	extra := shardCount % total
+	var lo, hi int
+	if idx < extra {
+		lo = idx * (base + 1)
+		hi = lo + base + 1
+	} else {
+		lo = idx*base + extra
+		hi = lo + base
+	}
+	if hi > shardCount {
+		hi = shardCount
+	}
+	if lo >= hi {
+		return nil
+	}
+	out := make([]int, 0, hi-lo)
+	for i := lo; i < hi; i++ {
+		out = append(out, i)
+	}
+	return out
+}

weed/s3api/s3lifecycle/scheduler/scheduler_test.go

@@ -0,0 +1,86 @@
+package scheduler
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestAssignShardsSingleWorker(t *testing.T) {
+	got := AssignShards(0, 1)
+	want := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestAssignShardsEvenSplit(t *testing.T) {
+	a := AssignShards(0, 2)
+	b := AssignShards(1, 2)
+	wantA := []int{0, 1, 2, 3, 4, 5, 6, 7}
+	wantB := []int{8, 9, 10, 11, 12, 13, 14, 15}
+	if !reflect.DeepEqual(a, wantA) {
+		t.Fatalf("worker 0: got %v, want %v", a, wantA)
+	}
+	if !reflect.DeepEqual(b, wantB) {
+		t.Fatalf("worker 1: got %v, want %v", b, wantB)
+	}
+}
+
+func TestAssignShardsUnevenSplit(t *testing.T) {
+	// 16 shards / 3 workers = 5,5,5 + 1 remainder. Earlier workers absorb
+	// the remainder so the slices stay contiguous: 6,5,5.
+	w0 := AssignShards(0, 3)
+	w1 := AssignShards(1, 3)
+	w2 := AssignShards(2, 3)
+	if len(w0) != 6 || w0[0] != 0 || w0[len(w0)-1] != 5 {
+		t.Fatalf("w0 expected [0..5], got %v", w0)
+	}
+	if len(w1) != 5 || w1[0] != 6 || w1[len(w1)-1] != 10 {
+		t.Fatalf("w1 expected [6..10], got %v", w1)
+	}
+	if len(w2) != 5 || w2[0] != 11 || w2[len(w2)-1] != 15 {
+		t.Fatalf("w2 expected [11..15], got %v", w2)
+	}
+	// All shards covered, exactly once.
+	covered := map[int]int{}
+	for _, w := range [][]int{w0, w1, w2} {
+		for _, s := range w {
+			covered[s]++
+		}
+	}
+	if len(covered) != 16 {
+		t.Fatalf("expected 16 unique shards, got %d", len(covered))
+	}
+	for s, c := range covered {
+		if c != 1 {
+			t.Fatalf("shard %d covered %d times, expected 1", s, c)
+		}
+	}
+}
+
+func TestAssignShardsMoreWorkersThanShards(t *testing.T) {
+	// 20 workers, 16 shards: first 16 get one shard each, last 4 get nothing.
+	for i := 0; i < 16; i++ {
+		got := AssignShards(i, 20)
+		if len(got) != 1 || got[0] != i {
+			t.Fatalf("worker %d: got %v, want [%d]", i, got, i)
+		}
+	}
+	for i := 16; i < 20; i++ {
+		if got := AssignShards(i, 20); got != nil {
+			t.Fatalf("worker %d should get nil, got %v", i, got)
+		}
+	}
+}
+
+func TestAssignShardsBounds(t *testing.T) {
+	if got := AssignShards(-1, 4); got != nil {
+		t.Fatalf("idx -1: got %v, want nil", got)
+	}
+	if got := AssignShards(4, 4); got != nil {
+		t.Fatalf("idx==total: got %v, want nil", got)
+	}
+	if got := AssignShards(0, 0); got != nil {
+		t.Fatalf("total 0: got %v, want nil", got)
+	}
+}

weed/shell/command_s3_lifecycle_run_shard.go

@@ -13,11 +13,10 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/lifecycle_xml"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
@@ -102,16 +101,13 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i
 	// Run the whole pipeline inside one WithFilerClient so the reader's
 	// SubscribeMetadata stream and the persister share a single connection.
 	return env.WithFilerClient(true, func(filerClient filer_pb.SeaweedFilerClient) error {
-		inputs, parseErrors, err := loadLifecycleCompileInputs(context.Background(), filerClient, bucketsPath)
+		inputs, parseErrors, err := scheduler.LoadCompileInputs(context.Background(), filerClient, bucketsPath)
 		if err != nil {
 			return fmt.Errorf("load lifecycle configs: %w", err)
 		}
 		for i, pe := range parseErrors {
-			// Surface up to the first three parse errors so the operator
-			// can chase malformed configs; cap the rest with a count so
-			// the output stays readable on large clusters.
 			if i < 3 {
-				fmt.Fprintf(writer, "warning: %s: %v\n", pe.bucket, pe.err)
+				fmt.Fprintf(writer, "warning: %s: %v\n", pe.Bucket, pe.Err)
 			}
 		}
 		if extra := len(parseErrors) - 3; extra > 0 {
@@ -123,11 +119,8 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i
 		}
 		fmt.Fprintf(writer, "loaded lifecycle for %d bucket(s)\n", len(inputs))
 
-		// Activate every action so this manual run dispatches whatever fires.
-		// The production bootstrap walker promotes actions only after a clean
-		// walk; this shell entrypoint runs out-of-band of that flow.
 		eng := engine.New()
-		eng.Compile(inputs, engine.CompileOptions{PriorStates: allActivePriorStates(inputs)})
+		eng.Compile(inputs, engine.CompileOptions{PriorStates: scheduler.AllActivePriorStates(inputs)})
 
 		pipeline := &dispatcher.Pipeline{
 			Shards:         shards,
@@ -288,89 +281,3 @@ func resolveBucketsPath(env *CommandEnv) (string, error) {
 	return path, nil
 }
 
-type lifecycleParseError struct {
-	bucket string
-	err    error
-}
-
-// loadLifecycleCompileInputs walks the buckets directory and reads each
-// bucket entry's lifecycle XML from its Extended attributes. Pagination
-// loops with startFrom so clusters with more than one page of buckets
-// don't drop the tail. Parse errors are collected per bucket and returned
-// alongside the successful inputs so the caller can surface them.
-func loadLifecycleCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketsPath string) ([]engine.CompileInput, []lifecycleParseError, error) {
-	var (
-		inputs      []engine.CompileInput
-		parseErrors []lifecycleParseError
-		startFrom   string
-	)
-	const pageSize uint32 = 1024
-	for {
-		pageCount := 0
-		var lastName string
-		err := filer_pb.SeaweedList(ctx, client, bucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
-			pageCount++
-			lastName = entry.Name
-			if !entry.IsDirectory {
-				return nil
-			}
-			xmlBytes, ok := entry.Extended[bucketLifecycleConfigurationXMLKey]
-			if !ok || len(xmlBytes) == 0 {
-				return nil
-			}
-			rules, err := lifecycle_xml.ParseCanonical(xmlBytes)
-			if err != nil {
-				parseErrors = append(parseErrors, lifecycleParseError{bucket: entry.Name, err: err})
-				return nil
-			}
-			if len(rules) == 0 {
-				return nil
-			}
-			inputs = append(inputs, engine.CompileInput{
-				Bucket:    entry.Name,
-				Rules:     rules,
-				Versioned: isBucketVersioned(entry),
-			})
-			return nil
-		}, startFrom, false, pageSize)
-		if err != nil {
-			return nil, nil, err
-		}
-		if uint32(pageCount) < pageSize {
-			break
-		}
-		startFrom = lastName
-	}
-	return inputs, parseErrors, nil
-}
-
-const bucketLifecycleConfigurationXMLKey = "s3-bucket-lifecycle-configuration-xml"
-
-func isBucketVersioned(entry *filer_pb.Entry) bool {
-	v, ok := entry.Extended[s3_constants.ExtVersioningKey]
-	if !ok {
-		return false
-	}
-	s := strings.ToLower(strings.TrimSpace(string(v)))
-	return s == "enabled" || s == "suspended"
-}
-
-// allActivePriorStates seeds every compiled action as bootstrap-complete +
-// event-driven so the run dispatches whatever fires. Production bootstrap
-// walks set this incrementally per bucket; this manual run skips the walk.
-func allActivePriorStates(inputs []engine.CompileInput) map[s3lifecycle.ActionKey]engine.PriorState {
-	prior := map[s3lifecycle.ActionKey]engine.PriorState{}
-	for _, in := range inputs {
-		for _, rule := range in.Rules {
-			hash := s3lifecycle.RuleHash(rule)
-			for _, kind := range s3lifecycle.RuleActionKinds(rule) {
-				key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: hash, ActionKind: kind}
-				prior[key] = engine.PriorState{
-					BootstrapComplete: true,
-					Mode:              engine.ModeEventDriven,
-				}
-			}
-		}
-	}
-	return prior
-}

weed/worker/tasks/s3_lifecycle/config.go

@@ -0,0 +1,70 @@
+package s3_lifecycle
+
+import (
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+const (
+	jobType = "s3_lifecycle"
+
+	defaultWorkers                = 1
+	defaultDispatchTickMinutes    = int64(1)
+	defaultCheckpointTickSeconds  = int64(30)
+	defaultRefreshIntervalMinutes = int64(5)
+	defaultMaxRuntimeMinutes      = int64(60)
+)
+
+// Config is the parsed AdminConfigForm + WorkerConfigForm view.
+type Config struct {
+	Workers         int
+	DispatchTick    time.Duration
+	CheckpointTick  time.Duration
+	RefreshInterval time.Duration
+	MaxRuntime      time.Duration
+}
+
+// ParseConfig pulls the lifecycle Handler config from the merged
+// admin+worker config values. Missing fields fall back to defaults.
+func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Config {
+	cfg := Config{
+		Workers:         int(readInt64(adminValues, "workers", defaultWorkers)),
+		DispatchTick:    time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute,
+		CheckpointTick:  time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second,
+		RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute,
+		MaxRuntime:      time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute,
+	}
+	if cfg.Workers <= 0 {
+		cfg.Workers = defaultWorkers
+	}
+	if cfg.DispatchTick <= 0 {
+		cfg.DispatchTick = time.Duration(defaultDispatchTickMinutes) * time.Minute
+	}
+	if cfg.CheckpointTick <= 0 {
+		cfg.CheckpointTick = time.Duration(defaultCheckpointTickSeconds) * time.Second
+	}
+	if cfg.RefreshInterval <= 0 {
+		cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute
+	}
+	if cfg.MaxRuntime <= 0 {
+		cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute
+	}
+	return cfg
+}
+
+func readInt64(values map[string]*plugin_pb.ConfigValue, field string, fallback int64) int64 {
+	v, ok := values[field]
+	if !ok || v == nil {
+		return fallback
+	}
+	switch k := v.Kind.(type) {
+	case *plugin_pb.ConfigValue_Int64Value:
+		return k.Int64Value
+	case *plugin_pb.ConfigValue_DoubleValue:
+		return int64(k.DoubleValue)
+	case *plugin_pb.ConfigValue_StringValue:
+		return fallback
+	}
+	return fallback
+}

weed/worker/tasks/s3_lifecycle/config_test.go

@@ -0,0 +1,77 @@
+package s3_lifecycle
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+func TestParseConfigDefaults(t *testing.T) {
+	cfg := ParseConfig(nil, nil)
+	if cfg.Workers != defaultWorkers {
+		t.Errorf("Workers default=%d, want %d", cfg.Workers, defaultWorkers)
+	}
+	if cfg.DispatchTick != 1*time.Minute {
+		t.Errorf("DispatchTick default=%v, want 1m", cfg.DispatchTick)
+	}
+	if cfg.CheckpointTick != 30*time.Second {
+		t.Errorf("CheckpointTick default=%v, want 30s", cfg.CheckpointTick)
+	}
+	if cfg.RefreshInterval != 5*time.Minute {
+		t.Errorf("RefreshInterval default=%v, want 5m", cfg.RefreshInterval)
+	}
+	if cfg.MaxRuntime != 60*time.Minute {
+		t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime)
+	}
+}
+
+func TestParseConfigOverrides(t *testing.T) {
+	admin := map[string]*plugin_pb.ConfigValue{
+		"workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}},
+	}
+	worker := map[string]*plugin_pb.ConfigValue{
+		"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}},
+		"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}},
+		"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}},
+		"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}},
+	}
+	cfg := ParseConfig(admin, worker)
+	if cfg.Workers != 4 {
+		t.Errorf("Workers=%d, want 4", cfg.Workers)
+	}
+	if cfg.DispatchTick != 2*time.Minute {
+		t.Errorf("DispatchTick=%v, want 2m", cfg.DispatchTick)
+	}
+	if cfg.CheckpointTick != 15*time.Second {
+		t.Errorf("CheckpointTick=%v, want 15s", cfg.CheckpointTick)
+	}
+	if cfg.RefreshInterval != 10*time.Minute {
+		t.Errorf("RefreshInterval=%v, want 10m", cfg.RefreshInterval)
+	}
+	if cfg.MaxRuntime != 120*time.Minute {
+		t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime)
+	}
+}
+
+func TestParseConfigClampsZeroAndNegative(t *testing.T) {
+	worker := map[string]*plugin_pb.ConfigValue{
+		"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
+		"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}},
+	}
+	cfg := ParseConfig(nil, worker)
+	if cfg.DispatchTick != 1*time.Minute {
+		t.Errorf("zero DispatchTick should clamp to default, got %v", cfg.DispatchTick)
+	}
+	if cfg.CheckpointTick != 30*time.Second {
+		t.Errorf("zero CheckpointTick should clamp to default, got %v", cfg.CheckpointTick)
+	}
+	if cfg.RefreshInterval != 5*time.Minute {
+		t.Errorf("zero RefreshInterval should clamp to default, got %v", cfg.RefreshInterval)
+	}
+	if cfg.MaxRuntime != 60*time.Minute {
+		t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime)
+	}
+}

weed/worker/tasks/s3_lifecycle/handler.go

@@ -0,0 +1,308 @@
+package s3_lifecycle
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	pluginworker.RegisterHandler(pluginworker.HandlerFactory{
+		JobType:  jobType,
+		Category: pluginworker.CategoryDefault,
+		Aliases:  []string{"s3-lifecycle", "s3.lifecycle", "lifecycle"},
+		Build: func(opts pluginworker.HandlerBuildOptions) (pluginworker.JobHandler, error) {
+			return NewHandler(opts.GrpcDialOption), nil
+		},
+	})
+}
+
+// Handler is the worker-side runner for S3 object lifecycle expiration.
+// One Execute call drives a long-running scheduler.Scheduler against the
+// S3 endpoints discovered from the master; admin caps concurrency at one
+// job per worker so a fresh proposal only spawns a new run after the
+// prior one exits.
+type Handler struct {
+	grpcDialOption grpc.DialOption
+}
+
+func NewHandler(grpcDialOption grpc.DialOption) *Handler {
+	return &Handler{grpcDialOption: grpcDialOption}
+}
+
+func (h *Handler) Capability() *plugin_pb.JobTypeCapability {
+	return &plugin_pb.JobTypeCapability{
+		JobType:                 jobType,
+		CanDetect:               true,
+		CanExecute:              true,
+		MaxDetectionConcurrency: 1,
+		MaxExecutionConcurrency: 1,
+		DisplayName:             "S3 Lifecycle",
+		Description:             "Daily batch: scan the filer meta-log and delete objects whose lifecycle rule has fired.",
+		Weight:                  20,
+	}
+}
+
+func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
+	return &plugin_pb.JobTypeDescriptor{
+		JobType:           jobType,
+		DisplayName:       "S3 Lifecycle",
+		Description:       "Daily S3 object expiration scan with per-shard cursors and bounded retry.",
+		Icon:              "fas fa-recycle",
+		DescriptorVersion: 1,
+		AdminConfigForm: &plugin_pb.ConfigForm{
+			FormId:      "s3-lifecycle-admin",
+			Title:       "S3 Lifecycle",
+			Description: "Cluster-wide controls for the lifecycle scheduler.",
+			Sections: []*plugin_pb.ConfigSection{
+				{
+					SectionId:   "scope",
+					Title:       "Scope",
+					Description: "How many pipeline goroutines split the 16-shard space.",
+					Fields: []*plugin_pb.ConfigField{
+						{
+							Name:        "workers",
+							Label:       "Worker Count",
+							Description: "Number of pipeline goroutines per executing worker. Each owns a contiguous slice of [0, 16) shards. Default 1 = one goroutine handles all 16 shards.",
+							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}},
+							MaxValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 16}},
+						},
+					},
+				},
+			},
+			DefaultValues: map[string]*plugin_pb.ConfigValue{
+				"workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultWorkers}},
+			},
+		},
+		WorkerConfigForm: &plugin_pb.ConfigForm{
+			FormId:      "s3-lifecycle-worker",
+			Title:       "S3 Lifecycle Worker",
+			Description: "Operational tuning for the lifecycle pipeline.",
+			Sections: []*plugin_pb.ConfigSection{
+				{
+					SectionId:   "cadence",
+					Title:       "Cadence",
+					Description: "Tick intervals for the dispatch loop, durable checkpoint, and engine refresh; plus the wall-clock cap on each daily run.",
+					Fields: []*plugin_pb.ConfigField{
+						{
+							Name:        "dispatch_tick_minutes",
+							Label:       "Dispatch Tick (minutes)",
+							Description: "How often each pipeline drains its schedule.",
+							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}},
+						},
+						{
+							Name:        "checkpoint_tick_seconds",
+							Label:       "Cursor Checkpoint Tick (seconds)",
+							Description: "How often each pipeline persists its cursor map to the filer.",
+							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}},
+						},
+						{
+							Name:        "refresh_interval_minutes",
+							Label:       "Engine Refresh Interval (minutes)",
+							Description: "How often the scheduler rebuilds the engine snapshot from bucket lifecycle configs.",
+							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}},
+						},
+						{
+							Name:        "max_runtime_minutes",
+							Label:       "Max Runtime (minutes)",
+							Description: "Wall-clock cap on each run. Each daily run processes events for one day; 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{
+				"dispatch_tick_minutes":    {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}},
+				"checkpoint_tick_seconds":  {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}},
+				"refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}},
+				"max_runtime_minutes":      {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}},
+			},
+		},
+		AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
+			DetectionIntervalSeconds: 24 * 60 * 60, // daily
+			DetectionTimeoutSeconds:  60,
+			MaxJobsPerDetection:      1,
+		},
+	}
+}
+
+func (h *Handler) Detect(ctx context.Context, request *plugin_pb.RunDetectionRequest, sender pluginworker.DetectionSender) error {
+	if request == nil || sender == nil {
+		return fmt.Errorf("detect: nil request or sender")
+	}
+	if request.JobType != "" && request.JobType != jobType {
+		return fmt.Errorf("job type %q is not handled by s3 lifecycle handler", request.JobType)
+	}
+	s3Endpoints := clusterS3Endpoints(request.ClusterContext)
+	if len(s3Endpoints) == 0 {
+		_ = sender.SendActivity(pluginworker.BuildDetectorActivity("skipped", "no s3 servers registered with master", nil))
+		return sender.SendComplete(&plugin_pb.DetectionComplete{JobType: jobType, Success: true})
+	}
+	filerAddresses := []string{}
+	if request.ClusterContext != nil {
+		filerAddresses = append(filerAddresses, request.ClusterContext.FilerGrpcAddresses...)
+	}
+	if len(filerAddresses) == 0 {
+		_ = sender.SendActivity(pluginworker.BuildDetectorActivity("skipped", "no filer addresses in cluster context", nil))
+		return sender.SendComplete(&plugin_pb.DetectionComplete{JobType: jobType, Success: true})
+	}
+
+	proposal := &plugin_pb.JobProposal{
+		JobType:    jobType,
+		ProposalId: fmt.Sprintf("s3-lifecycle-%d", time.Now().UnixNano()),
+		Priority:   plugin_pb.JobPriority_JOB_PRIORITY_NORMAL,
+		Parameters: map[string]*plugin_pb.ConfigValue{
+			"filer_grpc_address": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: filerAddresses[0]}},
+		},
+	}
+	if err := sender.SendProposals(&plugin_pb.DetectionProposals{
+		JobType:   jobType,
+		Proposals: []*plugin_pb.JobProposal{proposal},
+	}); err != nil {
+		return err
+	}
+	return sender.SendComplete(&plugin_pb.DetectionComplete{
+		JobType:        jobType,
+		Success:        true,
+		TotalProposals: 1,
+	})
+}
+
+func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender) error {
+	if request == nil || request.Job == nil || sender == nil {
+		return fmt.Errorf("execute: nil request/job/sender")
+	}
+	if request.Job.JobType != "" && request.Job.JobType != jobType {
+		return fmt.Errorf("job type %q is not handled by s3 lifecycle handler", request.Job.JobType)
+	}
+	cfg := ParseConfig(request.GetAdminConfigValues(), request.GetWorkerConfigValues())
+	s3Endpoints := clusterS3Endpoints(request.ClusterContext)
+	if len(s3Endpoints) == 0 {
+		return fmt.Errorf("execute: no s3 servers registered with master")
+	}
+	filerAddress := readString(request.Job.Parameters, "filer_grpc_address", "")
+	if filerAddress == "" {
+		return fmt.Errorf("execute: missing filer_grpc_address in job parameters")
+	}
+
+	runCtx, cancel := context.WithTimeout(ctx, cfg.MaxRuntime)
+	defer cancel()
+
+	_ = 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),
+	})
+
+	dialCtx, dialCancel := context.WithTimeout(runCtx, 30*time.Second)
+	filerConn, err := pb.GrpcDial(dialCtx, filerAddress, false, h.grpcDialOption)
+	dialCancel()
+	if err != nil {
+		return fmt.Errorf("dial filer %s: %w", filerAddress, err)
+	}
+	defer filerConn.Close()
+	filerClient := filer_pb.NewSeaweedFilerClient(filerConn)
+
+	bucketsPath, err := lookupBucketsPath(runCtx, filerClient)
+	if err != nil {
+		return fmt.Errorf("buckets path: %w", err)
+	}
+
+	dialCtx, dialCancel = context.WithTimeout(runCtx, 30*time.Second)
+	s3Conn, err := pb.GrpcDial(dialCtx, s3Endpoints[0], false, h.grpcDialOption)
+	dialCancel()
+	if err != nil {
+		return fmt.Errorf("dial s3 %s: %w", s3Endpoints[0], err)
+	}
+	defer s3Conn.Close()
+	rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn)
+
+	sched := &scheduler.Scheduler{
+		BucketsPath:     bucketsPath,
+		Engine:          engine.New(),
+		Persister:       &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
+		Client:          lifecycleRPCAdapter{c: rpc},
+		FilerClient:     filerClient,
+		ClientID:        util.RandomInt32(),
+		ClientName:      "worker-s3-lifecycle",
+		Workers:         cfg.Workers,
+		DispatchTick:    cfg.DispatchTick,
+		CheckpointTick:  cfg.CheckpointTick,
+		RefreshInterval: cfg.RefreshInterval,
+	}
+	if err := sched.Run(runCtx); err != nil {
+		glog.Warningf("s3 lifecycle execute: %v", err)
+		return err
+	}
+	return nil
+}
+
+// clusterS3Endpoints returns the master-discovered S3 gRPC addresses for the
+// cluster. The handler dials the first reachable one; the master refreshes
+// the list on KeepConnected so a stale entry self-heals on the next run.
+func clusterS3Endpoints(cc *plugin_pb.ClusterContext) []string {
+	if cc == nil {
+		return nil
+	}
+	out := make([]string, 0, len(cc.S3GrpcAddresses))
+	for _, addr := range cc.S3GrpcAddresses {
+		if addr != "" {
+			out = append(out, addr)
+		}
+	}
+	return out
+}
+
+type lifecycleRPCAdapter struct {
+	c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient
+}
+
+func (a lifecycleRPCAdapter) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+	return a.c.LifecycleDelete(ctx, req)
+}
+
+func lookupBucketsPath(ctx context.Context, client filer_pb.SeaweedFilerClient) (string, error) {
+	resp, err := client.GetFilerConfiguration(ctx, &filer_pb.GetFilerConfigurationRequest{})
+	if err != nil {
+		return "", err
+	}
+	if path := resp.GetDirBuckets(); path != "" {
+		return path, nil
+	}
+	return "/buckets", nil
+}
+
+func readString(values map[string]*plugin_pb.ConfigValue, field, fallback string) string {
+	v, ok := values[field]
+	if !ok || v == nil {
+		return fallback
+	}
+	if k, ok := v.Kind.(*plugin_pb.ConfigValue_StringValue); ok {
+		return k.StringValue
+	}
+	return fallback
+}
+
+var _ pluginworker.JobHandler = (*Handler)(nil)