From d5372f9eb7822fdeec88cdd85114ee339009286b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 13 May 2026 09:24:50 -0700 Subject: [PATCH] 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. --- .../s3lifecycle/dailyrun/walker_dispatcher.go | 14 ++++++++ .../dailyrun/walker_dispatcher_test.go | 34 +++++++++++++++++++ weed/worker/tasks/s3_lifecycle/handler.go | 4 ++- 3 files changed, 51 insertions(+), 1 deletion(-) diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go index d54e7bd81..eaff959b8 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go index 68c067cd6..f30239394 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go +++ b/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"})) diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 5fe2fa0d1..947036036 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/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) })