feat(s3/lifecycle): apply cluster rate limit to walker dispatch (#9471)

Phase 4b shipped the walker without plugging it into the cluster
rate.Limiter that processMatches honors. A walker hitting a large
bucket on the recovery branch could burst LifecycleDelete RPCs past
the cluster_deletes_per_second cap that streaming-replay respects.

WalkerDispatcher now takes a *rate.Limiter and waits on it before
each RPC, observing the wait time on S3LifecycleDispatchLimiterWaitSeconds
just like processMatches does. The handler passes the same limiter
to both paths so replay + walk share one budget; nil disables
throttling (unchanged default).

Tests pin: the limiter actually delays a dispatch when the burst
token is drained, and a ctx cancellation in Limiter.Wait surfaces
as an error without sending the RPC.

weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go

@@ -3,11 +3,13 @@ package dailyrun
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"golang.org/x/time/rate"
 )
 
 // WalkerDispatcher adapts LifecycleClient to bootstrap.Dispatcher so
@@ -18,6 +20,11 @@ import (
 // right contract for a full-tree walk that has just observed the entry.
 type WalkerDispatcher struct {
 	Client LifecycleClient
+	// Limiter throttles LifecycleDelete dispatch under the cluster's
+	// shared deletes/sec budget. Should be the same *rate.Limiter the
+	// daily-run's processMatches uses so the walker and replay paths
+	// can't combine to burst past the cap. nil disables throttling.
+	Limiter *rate.Limiter
 }
 
 // Compile-time check.
@@ -59,6 +66,13 @@ func (d *WalkerDispatcher) Delete(ctx context.Context, action *engine.CompiledAc
 		// the live entry on this code path.
 	}
 	kindLabel := action.Key.ActionKind.String()
+	if d.Limiter != nil {
+		waitStart := time.Now()
+		if waitErr := d.Limiter.Wait(ctx); waitErr != nil {
+			return fmt.Errorf("walker dispatch %s/%s %s: limiter: %w", action.Bucket, objectPath, action.Key.ActionKind, waitErr)
+		}
+		stats.S3LifecycleDispatchLimiterWaitSeconds.Observe(time.Since(waitStart).Seconds())
+	}
 	resp, err := d.Client.LifecycleDelete(ctx, req)
 	if err != nil {
 		// "RPC_ERROR" matches the streaming dispatcher and processMatches

weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
@@ -11,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
 )
 
 // walkerStubClient captures the last LifecycleDeleteRequest so tests
@@ -151,6 +153,38 @@ func TestWalkerDispatcher_NilResponseReturnsError(t *testing.T) {
 	assert.Contains(t, err.Error(), "nil response")
 }
 
+func TestWalkerDispatcher_LimiterWaitsBeforeDispatch(t *testing.T) {
+	// Build a tiny limiter (1 token, slow refill) and pre-drain it so
+	// the next Wait blocks until the deadline. Dispatcher must respect
+	// the limiter — without the wait the test passes trivially.
+	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
+	lim := rate.NewLimiter(rate.Every(50*time.Millisecond), 1)
+	_ = lim.AllowN(time.Now(), 1) // burn the burst token
+	d := &WalkerDispatcher{Client: c, Limiter: lim}
+
+	start := time.Now()
+	err := d.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})
+	elapsed := time.Since(start)
+	require.NoError(t, err)
+	assert.GreaterOrEqual(t, elapsed, 30*time.Millisecond,
+		"limiter must throttle the dispatch; elapsed=%v", elapsed)
+}
+
+func TestWalkerDispatcher_LimiterContextCancelHaltsWalker(t *testing.T) {
+	// Pre-drained limiter + canceled ctx. Limiter.Wait returns the
+	// cancel error; walker must surface it (not silently dispatch).
+	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
+	lim := rate.NewLimiter(rate.Every(time.Hour), 1)
+	_ = lim.AllowN(time.Now(), 1)
+	d := &WalkerDispatcher{Client: c, Limiter: lim}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	err := d.Delete(ctx, sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})
+	require.Error(t, err)
+	assert.Nil(t, c.lastReq, "no RPC should be sent when ctx was cancelled in Wait")
+}
+
 func TestWalkerDispatcher_NilGuardsReturnError(t *testing.T) {
 	d := &WalkerDispatcher{Client: &walkerStubClient{}}
 	require.Error(t, d.Delete(context.Background(), nil, &bootstrap.Entry{Path: "obj"}))

weed/worker/tasks/s3_lifecycle/handler.go

@@ -269,7 +269,9 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe
 		}
 	}
 	walkerListFn := dailyrun.FilerListFunc(filerClient, bucketsPath)
-	walkerDispatch := &dailyrun.WalkerDispatcher{Client: client}
+	// Share the limiter with processMatches so walker + replay can't
+	// combine to burst past the cluster cap.
+	walkerDispatch := &dailyrun.WalkerDispatcher{Client: client, Limiter: limiter}
 	walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shardID int) error {
 		return dailyrun.WalkBuckets(walkCtx, view, shardID, buckets, walkerListFn, walkerDispatch)
 	})