Revert "chore(s3/lifecycle): trim verbose comments"

This reverts commit 11b98ee873.

weed/admin/plugin/cluster_rate_limit.go

@@ -7,9 +7,23 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 )
 
-// String constants must match weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go.
-// Duplicated rather than imported so the admin package doesn't depend
-// on the worker handler package; tests pin both sides to the same values.
+// Job types whose ExecuteJobRequest needs per-worker rate-allocation
+// metadata injected. Keyed by the job-type string so plugin.go's
+// generic dispatch path stays job-agnostic.
+//
+// To add a new job type to the share-allocation pipeline: register an
+// entry here that knows how to read its admin-config field(s) and
+// produce the metadata keys/values the worker reads.
+
+// s3LifecycleClusterDeletesPerSecondKey, s3LifecycleClusterDeletesBurstKey,
+// s3LifecycleMetadataDeletesPerSecond, and s3LifecycleMetadataDeletesBurst
+// are the contract between admin and worker. The values must match the
+// constants exported from weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go
+// — duplicated here as plain strings rather than imported so the admin
+// plugin package doesn't pull a dependency on the worker handler
+// package. A mismatch on either side would silently disable rate
+// limiting; tests pin the constants in both packages against the same
+// values.
 const (
 	s3LifecycleJobType                    = "s3_lifecycle"
 	s3LifecycleClusterDeletesPerSecondKey = "cluster_deletes_per_second"
@@ -18,10 +32,23 @@ const (
 	s3LifecycleMetadataDeletesBurst       = "s3_lifecycle.deletes_burst"
 )
 
-// decorateClusterContextForJob injects per-job rate-allocation metadata
-// when the job type opts in. Divisor is min(executors, maxJobsPerDetection)
-// so singleton jobs (maxJobs=1) get the full cluster budget on the
-// single active worker.
+// decorateClusterContextForJob returns a new ClusterContext with any
+// per-job-type metadata the admin needs to inject before the
+// ExecuteJobRequest is sent. Returns the input cc unchanged when no
+// decoration applies.
+//
+// Today only s3_lifecycle decorates; the function exists so a future
+// job type's plumbing slots in alongside without touching
+// executeJobWithExecutor.
+//
+// maxJobsPerDetection is the job-type's AdminRuntimeConfig.MaxJobsPerDetection
+// — the cap on how many parallel instances of this job the scheduler will
+// dispatch per detection cycle. For singleton jobs (s3_lifecycle has
+// MaxJobsPerDetection=1) only one worker is ever active, so the cluster
+// budget must go to that worker undivided. For parallel-dispatch jobs the
+// budget divides across the actually-running set, not across every
+// capable worker. The divisor is min(executors, maxJobsPerDetection),
+// clamped to ≥1.
 func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobType string, adminConfigValues map[string]*plugin_pb.ConfigValue, maxJobsPerDetection int) *plugin_pb.ClusterContext {
 	if cc == nil {
 		return cc
@@ -32,13 +59,21 @@ func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobT
 	rps := readNonNegativeInt(adminConfigValues, s3LifecycleClusterDeletesPerSecondKey)
 	burst := readNonNegativeInt(adminConfigValues, s3LifecycleClusterDeletesBurstKey)
 	if rps <= 0 {
+		// Operator hasn't configured a cluster cap; nothing to allocate.
+		// The worker treats missing metadata keys as "unlimited," which
+		// is the legacy behavior.
 		return cc
 	}
 	executors := r.registry.CountCapableExecutors(jobType)
 	if executors <= 0 {
+		// No executors means the job won't dispatch at all; metadata
+		// would be discarded. Log so the case is visible in ops.
 		glog.V(2).Infof("decorateClusterContext: %s rps=%d but no execute-capable workers; skipping allocation", jobType, rps)
 		return cc
 	}
+	// Divide by the number of *concurrently-running* workers, not the
+	// number of capable ones. A singleton job (maxJobs=1) gets the full
+	// budget on its single active worker.
 	activeWorkers := executors
 	if maxJobsPerDetection > 0 && maxJobsPerDetection < activeWorkers {
 		activeWorkers = maxJobsPerDetection
@@ -48,11 +83,12 @@ func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobT
 	if burst > 0 {
 		perWorkerBurst = burst / activeWorkers
 		if perWorkerBurst < 1 {
-			// rate.Limiter with burst<1 never refills; floor.
 			perWorkerBurst = 1
 		}
 	}
 
+	// Clone so we don't mutate the shared base context. The metadata
+	// map is small; a fresh allocation per ExecuteJob is fine.
 	out := cloneClusterContext(cc)
 	if out.Metadata == nil {
 		out.Metadata = map[string]string{}
@@ -66,8 +102,11 @@ func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobT
 	return out
 }
 
-// cloneClusterContext duplicates the Metadata map so callers can mutate
-// it without racing the shared base context. Address slices are aliased.
+// cloneClusterContext returns a shallow-but-safe copy: the top-level
+// fields are reassigned, and the Metadata map is duplicated so the
+// caller can mutate it without racing other consumers of the input.
+// Slices of strings (master/filer/volume/s3 addresses) are copied by
+// reference — those are treated as immutable elsewhere in the codebase.
 func cloneClusterContext(in *plugin_pb.ClusterContext) *plugin_pb.ClusterContext {
 	if in == nil {
 		return nil
@@ -87,6 +126,9 @@ func cloneClusterContext(in *plugin_pb.ClusterContext) *plugin_pb.ClusterContext
 	return out
 }
 
+// readNonNegativeInt reads an int64 admin config value, treating
+// missing fields and non-int kinds as 0. Negative values are clamped
+// to 0 since the AdminConfigForm declares MinValue=0 on both fields.
 func readNonNegativeInt(values map[string]*plugin_pb.ConfigValue, field string) int {
 	v, ok := values[field]
 	if !ok || v == nil {

weed/admin/plugin/cluster_rate_limit_test.go

@@ -9,8 +9,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// pluginWithExecutors builds a Plugin with n execute-capable workers
-// for jobType, bypassing UpsertFromHello.
+// pluginWithExecutors returns a Plugin whose registry contains n
+// non-stale execute-capable workers for jobType. Helper for the
+// allocator tests. Bypasses UpsertFromHello so tests don't have to
+// build a full Hello message.
 func pluginWithExecutors(t *testing.T, jobType string, n int) *Plugin {
 	t.Helper()
 	reg := NewRegistry()
@@ -29,6 +31,7 @@ func pluginWithExecutors(t *testing.T, jobType string, n int) *Plugin {
 	return &Plugin{registry: reg}
 }
 
+// adminConfig builds an int64 admin config map for the given fields.
 func adminConfig(pairs ...interface{}) map[string]*plugin_pb.ConfigValue {
 	if len(pairs)%2 != 0 {
 		panic("adminConfig expects key/value pairs")
@@ -51,6 +54,9 @@ func adminConfig(pairs ...interface{}) map[string]*plugin_pb.ConfigValue {
 }
 
 func TestDecorateClusterContext_NonS3LifecycleIsPassThrough(t *testing.T) {
+	// Any job type other than s3_lifecycle gets the input cc back
+	// unchanged. Future allocators add their own branch; the default
+	// is pass-through.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{Metadata: map[string]string{"unrelated": "v"}}
 	out := r.decorateClusterContextForJob(in, "some_other_job", adminConfig(s3LifecycleClusterDeletesPerSecondKey, 100), 1)
@@ -58,7 +64,10 @@ func TestDecorateClusterContext_NonS3LifecycleIsPassThrough(t *testing.T) {
 }
 
 func TestDecorateClusterContext_RpsZeroSkipsAllocation(t *testing.T) {
-	// rps=0 must NOT write "0" — worker would read it as zero-throughput.
+	// rps=0 means "operator hasn't configured a cap"; the worker
+	// treats missing keys as unlimited. We must NOT inject any
+	// metadata (in particular, not "0") because that would force the
+	// worker into a no-throughput state on a misconfigured cluster.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{}
 	out := r.decorateClusterContextForJob(in, s3LifecycleJobType, adminConfig(s3LifecycleClusterDeletesPerSecondKey, 0), 1)
@@ -79,26 +88,34 @@ func TestDecorateClusterContext_NoExecutorsSkipsAllocation(t *testing.T) {
 }
 
 func TestDecorateClusterContext_SingletonJobGetsFullBudget(t *testing.T) {
-	// maxJobs=1: budget must go undivided to the single active worker.
+	// s3_lifecycle has MaxJobsPerDetection=1: only ONE worker runs the
+	// job at a time. The cluster budget must go to that worker undivided
+	// — dividing by N capable executors would starve the active worker
+	// to 1/N of the configured rps. Pin the singleton behavior.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{}
 	out := r.decorateClusterContextForJob(in, s3LifecycleJobType,
 		adminConfig(s3LifecycleClusterDeletesPerSecondKey, 100), 1)
 	require.NotNil(t, out.Metadata)
-	assert.Equal(t, "100", out.Metadata[s3LifecycleMetadataDeletesPerSecond])
+	assert.Equal(t, "100", out.Metadata[s3LifecycleMetadataDeletesPerSecond], "singleton job: full budget to the single active worker")
 }
 
 func TestDecorateClusterContext_SharedEvenlyWhenParallelLimited(t *testing.T) {
+	// Hypothetical parallel-dispatch job type (maxJobs=4): budget
+	// divides across the running-set, which equals min(executors,
+	// maxJobs)=4. 100/4=25.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{}
 	out := r.decorateClusterContextForJob(in, s3LifecycleJobType,
 		adminConfig(s3LifecycleClusterDeletesPerSecondKey, 100), 4)
 	require.NotNil(t, out.Metadata)
-	assert.Equal(t, "25", out.Metadata[s3LifecycleMetadataDeletesPerSecond])
+	assert.Equal(t, "25", out.Metadata[s3LifecycleMetadataDeletesPerSecond], "maxJobs=4 across 4 executors = 25/worker")
 }
 
 func TestDecorateClusterContext_MaxJobsExceedsExecutors(t *testing.T) {
-	// divisor = min(executors=4, maxJobs=10).
+	// maxJobs=10 but only 4 executors exist — the divisor is the
+	// smaller value (executors) since you can't run more jobs in
+	// parallel than there are workers to run them.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{}
 	out := r.decorateClusterContextForJob(in, s3LifecycleJobType,
@@ -118,7 +135,9 @@ func TestDecorateClusterContext_BurstSharedWhenParallel(t *testing.T) {
 }
 
 func TestDecorateClusterContext_BurstZeroOmitsKey(t *testing.T) {
-	// burst=0 means "let the worker default to 2*rps" — omit the key.
+	// burst=0 means "let the worker default it." Don't write the key —
+	// the worker's parsePositiveInt would then take the unset path
+	// and compute 2 × rps automatically.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{}
 	out := r.decorateClusterContextForJob(in, s3LifecycleJobType,
@@ -128,7 +147,8 @@ func TestDecorateClusterContext_BurstZeroOmitsKey(t *testing.T) {
 }
 
 func TestDecorateClusterContext_BurstFloorIsOneWhenDividesBelowOne(t *testing.T) {
-	// rate.Limiter with burst<1 never refills.
+	// burst=1 across 4 active workers would round to 0; clamp to 1 so
+	// the limiter doesn't become "single-token bucket that never refills."
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	in := &plugin_pb.ClusterContext{}
 	out := r.decorateClusterContextForJob(in, s3LifecycleJobType,
@@ -137,7 +157,9 @@ func TestDecorateClusterContext_BurstFloorIsOneWhenDividesBelowOne(t *testing.T)
 }
 
 func TestDecorateClusterContext_DoesNotMutateInput(t *testing.T) {
-	// Base ClusterContext is shared across parallel ExecuteJob calls.
+	// The same base ClusterContext is shared across many parallel
+	// ExecuteJob calls. The decorator must produce a fresh map so it
+	// can't race / leak per-job metadata into the base.
 	r := pluginWithExecutors(t, s3LifecycleJobType, 4)
 	baseMeta := map[string]string{"existing": "value"}
 	in := &plugin_pb.ClusterContext{Metadata: baseMeta}

weed/admin/plugin/plugin.go

@@ -649,6 +649,12 @@ func (r *Plugin) executeJobWithExecutor(
 		return nil, err
 	}
 
+	// Apply per-job-type cluster-allocation decoration (e.g. s3_lifecycle
+	// divides cluster_deletes_per_second by min(workers, maxJobsPerDetection)
+	// and ships the share via ClusterContext.Metadata). No-op for job
+	// types without an allocator registered. MaxJobsPerDetection caps
+	// the divisor so a singleton job (maxJobs=1) gets the full budget on
+	// the single active worker, not 1/N of it.
 	clusterContext = r.decorateClusterContextForJob(clusterContext, job.JobType, adminConfigValues, int(adminRuntime.GetMaxJobsPerDetection()))
 
 	completedCh := make(chan *plugin_pb.JobCompleted, 1)

weed/admin/plugin/registry.go

@@ -143,7 +143,11 @@ func (r *Registry) HasCapableWorker(jobType string) bool {
 }
 
 // CountCapableExecutors returns the number of non-stale workers that
-// can EXECUTE the given job type.
+// can EXECUTE the given job type. Used by per-job-type cluster
+// allocators (e.g. the s3_lifecycle delete-rate divider) to compute a
+// per-worker share at dispatch time. Returns 0 when no executor is
+// available — callers should treat that as "skip allocation" rather
+// than dividing by zero.
 func (r *Registry) CountCapableExecutors(jobType string) int {
 	r.mu.RLock()
 	defer r.mu.RUnlock()