feat: park scheduled jobs on per-worker admin queues

Assign a detection batch to the least-loaded replica-host worker in
one pass, then let ExecutionConcurrency goroutines pop a home-runnable
job or steal when that home queue reaches K. Dispatch no longer holds
a job in reserve-backoff, so a free worker is not blocked by another
home's pin. Tracked jobs stay pending/queued and record the home
worker for the admin UI.

weed/admin/plugin/plugin_monitor.go

@@ -887,6 +887,37 @@ func (r *Plugin) trackExecutionQueued(job *plugin_pb.JobSpec) {
 	})
 }
 
+// bindQueuedJobWorker records the home worker while the job stays pending/queued.
+func (r *Plugin) bindQueuedJobWorker(job *plugin_pb.JobSpec, workerID string) {
+	if job == nil || strings.TrimSpace(job.JobId) == "" || strings.TrimSpace(workerID) == "" {
+		return
+	}
+
+	now := time.Now().UTC()
+	r.jobsMu.Lock()
+	tracked := r.jobs[job.JobId]
+	if tracked == nil {
+		r.jobsMu.Unlock()
+		return
+	}
+	if isTerminalTrackedJobState(tracked.State) {
+		r.jobsMu.Unlock()
+		return
+	}
+	tracked.WorkerID = workerID
+	tracked.Message = "queued on worker " + workerID
+	tracked.UpdatedAt = timeToPtr(now)
+	snap := cloneTrackedJob(*tracked)
+	r.dirtyJobs = true
+	r.jobsMu.Unlock()
+
+	r.persistJobDetailSnapshot(job.JobId, func(detail *TrackedJob) {
+		detail.WorkerID = snap.WorkerID
+		detail.Message = snap.Message
+		detail.UpdatedAt = snap.UpdatedAt
+	})
+}
+
 func (r *Plugin) cancelQueuedJob(job *plugin_pb.JobSpec, cause error) {
 	reason := "job canceled"
 	if cause != nil {

weed/admin/plugin/plugin_scheduler.go

@@ -31,13 +31,13 @@ const (
 	defaultScheduledPerWorkerConcurrency       = 1
 	// Hold a job on replica-host workers until their reservation load
 	// reaches this; 0 disables the pin and spills immediately.
-	defaultLocalQueuePin                       = 5
-	maxScheduledExecutionConcurrency           = 128
-	defaultScheduledRetryBackoff               = 5 * time.Second
-	defaultClusterContextTimeout               = 10 * time.Second
-	defaultWaitingBacklogFloor                 = 8
-	defaultWaitingBacklogMultiplier            = 4
-	maxEstimatedRuntimeCap                     = 8 * time.Hour
+	defaultLocalQueuePin             = 5
+	maxScheduledExecutionConcurrency = 128
+	defaultScheduledRetryBackoff     = 5 * time.Second
+	defaultClusterContextTimeout     = 10 * time.Second
+	defaultWaitingBacklogFloor       = 8
+	defaultWaitingBacklogMultiplier  = 4
+	maxEstimatedRuntimeCap           = 8 * time.Hour
 	// How long started jobs may drain past the run window close.
 	scheduledExecutionDrainGrace = 30 * time.Minute
 )
@@ -824,19 +824,22 @@ func (r *Plugin) dispatchScheduledProposals(
 	// whole dispatch lets manual shell operations interleave between jobs.
 	jobsRequireLock := LaneRequiresLock(JobTypeLane(jobType))
 
-	jobQueue := make(chan *plugin_pb.JobSpec, len(proposals))
+	jobs := make([]*plugin_pb.JobSpec, 0, len(proposals))
 	for index, proposal := range proposals {
-		job := buildScheduledJobSpec(jobType, proposal, index)
-		r.trackExecutionQueued(job)
 		select {
 		case <-r.shutdownCh:
-			close(jobQueue)
-			return 0, 0, 0
+			for _, job := range jobs {
+				r.cancelQueuedJob(job, errSchedulerShutdown)
+			}
+			return 0, 0, len(jobs)
 		default:
-			jobQueue <- job
 		}
+		job := buildScheduledJobSpec(jobType, proposal, index)
+		r.trackExecutionQueued(job)
+		jobs = append(jobs, job)
 	}
-	close(jobQueue)
+
+	queues := r.assignJobsToWorkerQueues(jobType, jobs, policy)
 
 	var wg sync.WaitGroup
 	var statsMu sync.Mutex
@@ -853,130 +856,110 @@ func (r *Plugin) dispatchScheduledProposals(
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-
-		jobLoop:
-			for job := range jobQueue {
+			for {
 				select {
 				case <-r.shutdownCh:
 					return
 				default:
 				}
-
 				if ctx.Err() != nil {
-					r.cancelQueuedJob(job, ctx.Err())
-					statsMu.Lock()
-					canceledCount++
-					statsMu.Unlock()
+					return
+				}
+				if queues.isEmpty() {
+					return
+				}
+
+				item, targetID := r.takeQueuedJob(queues, jobType, policy)
+				if item == nil {
+					if queues.isEmpty() {
+						return
+					}
+					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
+						return
+					}
+					continue
+				}
+				if ctx.Err() != nil {
+					queues.pushFront(item)
+					return
+				}
+
+				executor, release, reserveErr := r.reserveExecutorByID(jobType, policy, targetID)
+				if reserveErr != nil {
+					queues.pushFront(item)
+					if ctx.Err() != nil {
+						return
+					}
+					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
+						return
+					}
 					continue
 				}
 
-				for {
-					select {
-					case <-r.shutdownCh:
-						return
-					default:
+				var releaseJobLock func()
+				if jobsRequireLock {
+					var lockErr error
+					releaseJobLock, lockErr = r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s job %s", jobType, item.job.JobId))
+					if lockErr != nil {
+						release()
+						statsMu.Lock()
+						errorCount++
+						statsMu.Unlock()
+						r.appendActivity(JobActivity{
+							JobID:      item.job.JobId,
+							JobType:    item.job.JobType,
+							Source:     "admin_scheduler",
+							Message:    fmt.Sprintf("scheduled execution lock failed: %v", lockErr),
+							Stage:      "failed",
+							OccurredAt: timeToPtr(time.Now().UTC()),
+						})
+						continue
 					}
 					if ctx.Err() != nil {
-						r.cancelQueuedJob(job, ctx.Err())
+						releaseJobLock()
+						release()
+						queues.pushFront(item)
+						return
+					}
+				}
+
+				err := r.executeScheduledJobWithExecutor(ctx, executor, item.job, clusterContext, policy)
+				if releaseJobLock != nil {
+					releaseJobLock()
+				}
+				release()
+				if errors.Is(err, errExecutorAtCapacity) {
+					queues.pushFront(item)
+					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
+						return
+					}
+					continue
+				}
+				if err != nil {
+					if ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+						r.cancelQueuedJob(item.job, err)
 						statsMu.Lock()
 						canceledCount++
 						statsMu.Unlock()
-						continue jobLoop
-					}
-
-					executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy, JobReplicaHosts(job))
-					if reserveErr != nil {
-						if ctx.Err() != nil {
-							r.cancelQueuedJob(job, ctx.Err())
-							statsMu.Lock()
-							canceledCount++
-							statsMu.Unlock()
-							continue jobLoop
-						}
-						statsMu.Lock()
-						errorCount++
-						statsMu.Unlock()
-						r.appendActivity(JobActivity{
-							JobType:    jobType,
-							Source:     "admin_scheduler",
-							Message:    fmt.Sprintf("scheduled execution reservation failed: %v", reserveErr),
-							Stage:      "failed",
-							OccurredAt: timeToPtr(time.Now().UTC()),
-						})
-						break
-					}
-
-					var releaseJobLock func()
-					if jobsRequireLock {
-						var lockErr error
-						releaseJobLock, lockErr = r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s job %s", jobType, job.JobId))
-						if lockErr != nil {
-							release()
-							statsMu.Lock()
-							errorCount++
-							statsMu.Unlock()
-							r.appendActivity(JobActivity{
-								JobID:      job.JobId,
-								JobType:    job.JobType,
-								Source:     "admin_scheduler",
-								Message:    fmt.Sprintf("scheduled execution lock failed: %v", lockErr),
-								Stage:      "failed",
-								OccurredAt: timeToPtr(time.Now().UTC()),
-							})
-							break
-						}
-						// The lock may have been held by an operator for a
-						// while; re-check the execution window.
-						if ctx.Err() != nil {
-							releaseJobLock()
-							release()
-							r.cancelQueuedJob(job, ctx.Err())
-							statsMu.Lock()
-							canceledCount++
-							statsMu.Unlock()
-							continue jobLoop
-						}
-					}
-
-					err := r.executeScheduledJobWithExecutor(ctx, executor, job, clusterContext, policy)
-					if releaseJobLock != nil {
-						releaseJobLock()
-					}
-					release()
-					if errors.Is(err, errExecutorAtCapacity) {
-						r.trackExecutionQueued(job)
-						if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
-							return
-						}
 						continue
 					}
-					if err != nil {
-						if ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
-							r.cancelQueuedJob(job, err)
-							statsMu.Lock()
-							canceledCount++
-							statsMu.Unlock()
-							continue jobLoop
-						}
-						statsMu.Lock()
-						errorCount++
-						statsMu.Unlock()
-						r.appendActivity(JobActivity{
-							JobID:      job.JobId,
-							JobType:    job.JobType,
-							Source:     "admin_scheduler",
-							Message:    fmt.Sprintf("scheduled execution failed: %v", err),
-							Stage:      "failed",
-							OccurredAt: timeToPtr(time.Now().UTC()),
-						})
-						break
-					}
-
 					statsMu.Lock()
-					successCount++
+					errorCount++
 					statsMu.Unlock()
-					break
+					r.appendActivity(JobActivity{
+						JobID:      item.job.JobId,
+						JobType:    item.job.JobType,
+						Source:     "admin_scheduler",
+						Message:    fmt.Sprintf("scheduled execution failed: %v", err),
+						Stage:      "failed",
+						OccurredAt: timeToPtr(time.Now().UTC()),
+					})
+					continue
 				}
+
+				statsMu.Lock()
+				successCount++
+				statsMu.Unlock()
 			}
 		}()
 	}
@@ -987,8 +970,8 @@ func (r *Plugin) dispatchScheduledProposals(
 	if drainErr == nil {
 		drainErr = errSchedulerShutdown
 	}
-	for job := range jobQueue {
-		r.cancelQueuedJob(job, drainErr)
+	for _, item := range queues.drain() {
+		r.cancelQueuedJob(item.job, drainErr)
 		canceledCount++
 	}
 
@@ -1047,15 +1030,6 @@ func (r *Plugin) reserveScheduledExecutor(
 			if executor, release, ok := r.tryReserveFirst(pref, jobType, policy); ok {
 				return executor, release, nil
 			}
-			if policy.LocalQueuePin > 0 && loadOf(pref[0]) < policy.LocalQueuePin {
-				if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
-					if ctx.Err() != nil {
-						return nil, nil, ctx.Err()
-					}
-					return nil, nil, fmt.Errorf("plugin is shutting down")
-				}
-				continue
-			}
 		}
 
 		sortExecutorsByLoad(rest, loadOf)
@@ -1090,6 +1064,28 @@ func (r *Plugin) tryReserveFirst(executors []*WorkerSession, jobType string, pol
 	return nil, nil, false
 }
 
+func (r *Plugin) reserveExecutorByID(jobType string, policy schedulerPolicy, workerID string) (*WorkerSession, func(), error) {
+	workerID = strings.TrimSpace(workerID)
+	if workerID == "" {
+		return nil, nil, fmt.Errorf("executor id is empty")
+	}
+	executors, err := r.registry.ListExecutors(jobType)
+	if err != nil {
+		return nil, nil, err
+	}
+	for _, executor := range executors {
+		if executor == nil || executor.WorkerID != workerID {
+			continue
+		}
+		release, ok := r.tryReserveExecutorCapacity(executor, jobType, policy)
+		if !ok {
+			return nil, nil, errExecutorAtCapacity
+		}
+		return executor, release, nil
+	}
+	return nil, nil, fmt.Errorf("executor %s is not available for %s", workerID, jobType)
+}
+
 func (r *Plugin) executorReservationLoad(executor *WorkerSession, lane SchedulerLane) int {
 	if executor == nil {
 		return 0

weed/admin/plugin/plugin_scheduler_test.go

@@ -996,135 +996,6 @@ func TestLaneStatus_LifecycleNextDetectionSurvivesDefaultLanePrune(t *testing.T)
 	}
 }
 
-func TestReservePinsToPreferredWhileLoadBelowK(t *testing.T) {
-	t.Parallel()
-
-	pluginSvc, err := New(Options{})
-	if err != nil {
-		t.Fatalf("New: %v", err)
-	}
-	defer pluginSvc.Shutdown()
-
-	pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
-		WorkerId: "worker-local",
-		Address:  "10.0.0.1:9334",
-		Capabilities: []*plugin_pb.JobTypeCapability{
-			{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 1},
-		},
-	})
-	pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
-		WorkerId: "worker-remote",
-		Address:  "10.0.0.2:9334",
-		Capabilities: []*plugin_pb.JobTypeCapability{
-			{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 4},
-		},
-	})
-
-	policy := schedulerPolicy{
-		ExecutionTimeout:       time.Second,
-		PerWorkerConcurrency:   1,
-		ExecutorReserveBackoff: 5 * time.Millisecond,
-		LocalQueuePin:          5,
-	}
-	preferred := []string{"10.0.0.1:8080"}
-
-	first, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred)
-	if err != nil {
-		t.Fatalf("reserve local: %v", err)
-	}
-	defer release1()
-	if first.WorkerID != "worker-local" {
-		t.Fatalf("first reserve want worker-local, got %s", first.WorkerID)
-	}
-
-	type reserveResult struct {
-		id  string
-		err error
-	}
-	secondCh := make(chan reserveResult, 1)
-	go func() {
-		exec, release, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred)
-		id := ""
-		if exec != nil {
-			id = exec.WorkerID
-		}
-		if release != nil {
-			release()
-		}
-		secondCh <- reserveResult{id: id, err: reserveErr}
-	}()
-
-	select {
-	case result := <-secondCh:
-		t.Fatalf("expected pin wait, got worker=%s err=%v", result.id, result.err)
-	case <-time.After(40 * time.Millisecond):
-	}
-
-	release1()
-
-	select {
-	case result := <-secondCh:
-		if result.err != nil {
-			t.Fatalf("second reserve after release: %v", result.err)
-		}
-		if result.id != "worker-local" {
-			t.Fatalf("after pin wait want worker-local, got %s", result.id)
-		}
-	case <-time.After(200 * time.Millisecond):
-		t.Fatalf("second reserve did not complete after local release")
-	}
-}
-
-func TestReserveSpillsWhenPinDisabled(t *testing.T) {
-	t.Parallel()
-
-	pluginSvc, err := New(Options{})
-	if err != nil {
-		t.Fatalf("New: %v", err)
-	}
-	defer pluginSvc.Shutdown()
-
-	pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
-		WorkerId: "worker-local",
-		Address:  "10.0.0.1:9334",
-		Capabilities: []*plugin_pb.JobTypeCapability{
-			{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 1},
-		},
-	})
-	pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
-		WorkerId: "worker-remote",
-		Address:  "10.0.0.2:9334",
-		Capabilities: []*plugin_pb.JobTypeCapability{
-			{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 4},
-		},
-	})
-
-	policy := schedulerPolicy{
-		PerWorkerConcurrency:   1,
-		ExecutorReserveBackoff: time.Millisecond,
-		LocalQueuePin:          0,
-	}
-	preferred := []string{"10.0.0.1:8080"}
-
-	first, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred)
-	if err != nil {
-		t.Fatalf("reserve local: %v", err)
-	}
-	defer release1()
-	if first.WorkerID != "worker-local" {
-		t.Fatalf("first reserve want worker-local, got %s", first.WorkerID)
-	}
-
-	second, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred)
-	if err != nil {
-		t.Fatalf("expected immediate spill, got %v", err)
-	}
-	defer release2()
-	if second.WorkerID != "worker-remote" {
-		t.Fatalf("K=0 should spill to worker-remote, got %s", second.WorkerID)
-	}
-}
-
 func TestReserveWaterfillsPreferredHosts(t *testing.T) {
 	t.Parallel()
 

weed/admin/plugin/worker_job_queues.go

@@ -0,0 +1,235 @@
+package plugin
+
+import (
+	"sync"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+// queuedJob is one scheduled spec parked on a home worker's admin-side queue.
+type queuedJob struct {
+	job  *plugin_pb.JobSpec
+	home string
+}
+
+// workerJobQueues holds per-worker FIFOs for one dispatch batch.
+// Home assignment is done once up front; run goroutines only pop/steal.
+type workerJobQueues struct {
+	mu        sync.Mutex
+	q         map[string][]*queuedJob
+	leftover  []*queuedJob
+	remaining int
+}
+
+func newWorkerJobQueues() *workerJobQueues {
+	return &workerJobQueues{q: make(map[string][]*queuedJob)}
+}
+
+func (qs *workerJobQueues) push(home string, job *plugin_pb.JobSpec) {
+	if qs == nil || job == nil {
+		return
+	}
+	item := &queuedJob{job: job, home: home}
+	qs.mu.Lock()
+	defer qs.mu.Unlock()
+	qs.pushLocked(item, false)
+}
+
+func (qs *workerJobQueues) pushFront(item *queuedJob) {
+	if qs == nil || item == nil {
+		return
+	}
+	qs.mu.Lock()
+	defer qs.mu.Unlock()
+	qs.pushLocked(item, true)
+}
+
+func (qs *workerJobQueues) pushLocked(item *queuedJob, front bool) {
+	qs.remaining++
+	if item.home == "" {
+		if front {
+			qs.leftover = append([]*queuedJob{item}, qs.leftover...)
+			return
+		}
+		qs.leftover = append(qs.leftover, item)
+		return
+	}
+	if front {
+		qs.q[item.home] = append([]*queuedJob{item}, qs.q[item.home]...)
+		return
+	}
+	qs.q[item.home] = append(qs.q[item.home], item)
+}
+
+func (qs *workerJobQueues) isEmpty() bool {
+	if qs == nil {
+		return true
+	}
+	qs.mu.Lock()
+	defer qs.mu.Unlock()
+	return qs.remaining == 0
+}
+
+func (qs *workerJobQueues) queueLen(workerID string) int {
+	if qs == nil {
+		return 0
+	}
+	qs.mu.Lock()
+	defer qs.mu.Unlock()
+	return len(qs.q[workerID])
+}
+
+func (qs *workerJobQueues) drain() []*queuedJob {
+	if qs == nil {
+		return nil
+	}
+	qs.mu.Lock()
+	defer qs.mu.Unlock()
+	out := make([]*queuedJob, 0, qs.remaining)
+	for id, qq := range qs.q {
+		out = append(out, qq...)
+		qs.q[id] = nil
+	}
+	out = append(out, qs.leftover...)
+	qs.leftover = nil
+	qs.remaining = 0
+	return out
+}
+
+func popFront(qq []*queuedJob) (*queuedJob, []*queuedJob) {
+	item := qq[0]
+	if len(qq) == 1 {
+		return item, nil
+	}
+	return item, qq[1:]
+}
+
+// assignJobsToWorkerQueues parks each job on the least-loaded replica-host
+// worker (or any worker if none match). Single-threaded; no steal here.
+func (r *Plugin) assignJobsToWorkerQueues(
+	jobType string,
+	jobs []*plugin_pb.JobSpec,
+	_ schedulerPolicy,
+) *workerJobQueues {
+	qs := newWorkerJobQueues()
+	executors, err := r.registry.ListExecutors(jobType)
+	lane := JobTypeLane(jobType)
+	queued := make(map[string]int, len(executors))
+
+	for _, job := range jobs {
+		if job == nil {
+			continue
+		}
+		if err != nil || len(executors) == 0 {
+			qs.push("", job)
+			continue
+		}
+		pref, rest := partitionByPreferredHost(executors, JobReplicaHosts(job))
+		cands := pref
+		if len(cands) == 0 {
+			cands = rest
+		}
+		sortExecutorsByLoad(cands, func(e *WorkerSession) int {
+			return r.executorReservationLoad(e, lane) + queued[e.WorkerID]
+		})
+		home := cands[0].WorkerID
+		qs.push(home, job)
+		queued[home]++
+		r.bindQueuedJobWorker(job, home)
+	}
+	return qs
+}
+
+// takeQueuedJob pops a home-runnable job, or steals from a queue at/over K
+// onto a free worker. Returns nil when nothing can start now.
+func (r *Plugin) takeQueuedJob(
+	qs *workerJobQueues,
+	jobType string,
+	policy schedulerPolicy,
+) (item *queuedJob, targetID string) {
+	if qs == nil {
+		return nil, ""
+	}
+	executors, err := r.registry.ListExecutors(jobType)
+	if err != nil || len(executors) == 0 {
+		return nil, ""
+	}
+	lane := JobTypeLane(jobType)
+	pin := policy.LocalQueuePin
+
+	qs.mu.Lock()
+	defer qs.mu.Unlock()
+	if qs.remaining == 0 {
+		return nil, ""
+	}
+
+	var bestHome *WorkerSession
+	bestLoad := 0
+	for _, exec := range executors {
+		if len(qs.q[exec.WorkerID]) == 0 {
+			continue
+		}
+		limit := schedulerWorkerExecutionLimit(exec, jobType, policy)
+		if limit <= 0 {
+			continue
+		}
+		load := r.executorReservationLoad(exec, lane)
+		if load >= limit {
+			continue
+		}
+		if bestHome == nil || load < bestLoad || (load == bestLoad && exec.WorkerID < bestHome.WorkerID) {
+			bestHome = exec
+			bestLoad = load
+		}
+	}
+	if bestHome != nil {
+		id := bestHome.WorkerID
+		item, qs.q[id] = popFront(qs.q[id])
+		qs.remaining--
+		return item, id
+	}
+
+	var thief *WorkerSession
+	for _, exec := range executors {
+		limit := schedulerWorkerExecutionLimit(exec, jobType, policy)
+		if limit <= 0 {
+			continue
+		}
+		if r.executorReservationLoad(exec, lane) >= limit {
+			continue
+		}
+		if thief == nil || exec.WorkerID < thief.WorkerID {
+			thief = exec
+		}
+	}
+	if thief == nil {
+		return nil, ""
+	}
+
+	if len(qs.leftover) > 0 {
+		item, qs.leftover = popFront(qs.leftover)
+		qs.remaining--
+		return item, thief.WorkerID
+	}
+
+	victimID := ""
+	victimLen := 0
+	for id, qq := range qs.q {
+		if len(qq) == 0 {
+			continue
+		}
+		if pin > 0 && len(qq) < pin {
+			continue
+		}
+		if victimID == "" || len(qq) > victimLen || (len(qq) == victimLen && id < victimID) {
+			victimID = id
+			victimLen = len(qq)
+		}
+	}
+	if victimID == "" {
+		return nil, ""
+	}
+	item, qs.q[victimID] = popFront(qs.q[victimID])
+	qs.remaining--
+	return item, thief.WorkerID
+}

weed/admin/plugin/worker_job_queues_test.go

@@ -0,0 +1,231 @@
+package plugin
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+func replicaJob(id, source string) *plugin_pb.JobSpec {
+	return &plugin_pb.JobSpec{
+		JobId:     id,
+		JobType:   "erasure_coding",
+		Summary:   id,
+		DedupeKey: id,
+		Parameters: map[string]*plugin_pb.ConfigValue{
+			"source_server": {
+				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: source},
+			},
+		},
+	}
+}
+
+func registerECWorkers(t *testing.T, pluginSvc *Plugin, workers map[string]string) {
+	t.Helper()
+	for id, addr := range workers {
+		pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+			WorkerId: id,
+			Address:  addr,
+			Capabilities: []*plugin_pb.JobTypeCapability{
+				{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 8},
+			},
+		})
+	}
+}
+
+func TestAssignJobsWaterfillsPreferredWorkers(t *testing.T) {
+	t.Parallel()
+
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	registerECWorkers(t, pluginSvc, map[string]string{
+		"worker-a": "10.0.0.1:9334",
+		"worker-b": "10.0.0.2:9334",
+		"worker-c": "10.0.0.3:9334",
+		"worker-d": "10.0.0.4:9334",
+	})
+
+	jobs := []*plugin_pb.JobSpec{
+		replicaJob("j1", "10.0.0.1:8080"),
+		replicaJob("j2", "10.0.0.1:8080"),
+		replicaJob("j3", "10.0.0.1:8080"),
+		replicaJob("j4", "10.0.0.1:8080"),
+	}
+	// All four replica IPs listed so every worker is preferred.
+	for _, job := range jobs {
+		job.Parameters["replica_servers"] = &plugin_pb.ConfigValue{
+			Kind: &plugin_pb.ConfigValue_StringList{StringList: &plugin_pb.StringList{
+				Values: []string{"10.0.0.1:8080", "10.0.0.2:8080", "10.0.0.3:8080", "10.0.0.4:8080"},
+			}},
+		}
+	}
+
+	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, schedulerPolicy{LocalQueuePin: 5})
+	if qs.queueLen("worker-a")+qs.queueLen("worker-b")+qs.queueLen("worker-c")+qs.queueLen("worker-d") != 4 {
+		t.Fatalf("expected 4 assigned jobs")
+	}
+	for _, id := range []string{"worker-a", "worker-b", "worker-c", "worker-d"} {
+		if qs.queueLen(id) != 1 {
+			t.Fatalf("want 1 job on %s, got %d", id, qs.queueLen(id))
+		}
+	}
+}
+
+func TestAssignAndTakeDoesNotStarveOtherHomes(t *testing.T) {
+	t.Parallel()
+
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	registerECWorkers(t, pluginSvc, map[string]string{
+		"worker-local":  "10.0.0.1:9334",
+		"worker-remote": "10.0.0.2:9334",
+	})
+
+	jobs := []*plugin_pb.JobSpec{
+		replicaJob("local-1", "10.0.0.1:8080"),
+		replicaJob("local-2", "10.0.0.1:8080"),
+		replicaJob("remote-1", "10.0.0.2:8080"),
+	}
+	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5}
+	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
+	if qs.queueLen("worker-local") != 2 || qs.queueLen("worker-remote") != 1 {
+		t.Fatalf("homes local=%d remote=%d", qs.queueLen("worker-local"), qs.queueLen("worker-remote"))
+	}
+
+	_, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	_, release1, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, target1)
+	if err != nil {
+		t.Fatalf("reserve 1: %v", err)
+	}
+	defer release1()
+
+	item2, target2 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if item2 == nil || target2 != "worker-remote" {
+		t.Fatalf("second take should start remote home, got job=%v target=%s", item2, target2)
+	}
+}
+
+func TestTakeStealsOnlyWhenHomeQueueReachesK(t *testing.T) {
+	t.Parallel()
+
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	registerECWorkers(t, pluginSvc, map[string]string{
+		"worker-local":  "10.0.0.1:9334",
+		"worker-remote": "10.0.0.2:9334",
+	})
+
+	jobs := make([]*plugin_pb.JobSpec, 0, 6)
+	for i := 0; i < 6; i++ {
+		jobs = append(jobs, replicaJob(fmt.Sprintf("j%d", i), "10.0.0.1:8080"))
+	}
+	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5}
+	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
+	if qs.queueLen("worker-local") != 6 {
+		t.Fatalf("all jobs should home on local, got %d", qs.queueLen("worker-local"))
+	}
+
+	_, target := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if target != "worker-local" {
+		t.Fatalf("first take want local, got %s", target)
+	}
+	_, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local")
+	if err != nil {
+		t.Fatalf("reserve local: %v", err)
+	}
+	defer release()
+
+	// 5 still queued on local (>=K) and local has no slot → steal to remote.
+	stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if stolen == nil || stealTarget != "worker-remote" {
+		t.Fatalf("expected steal to remote, got job=%v target=%s", stolen, stealTarget)
+	}
+	if qs.queueLen("worker-local") != 4 {
+		t.Fatalf("after steal want 4 left on local, got %d", qs.queueLen("worker-local"))
+	}
+
+	// 4 < K: do not steal again.
+	if item, tgt := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy); item != nil {
+		t.Fatalf("pin should hold remaining jobs, got target=%s", tgt)
+	}
+}
+
+func TestTakeStealsImmediatelyWhenPinDisabled(t *testing.T) {
+	t.Parallel()
+
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	registerECWorkers(t, pluginSvc, map[string]string{
+		"worker-local":  "10.0.0.1:9334",
+		"worker-remote": "10.0.0.2:9334",
+	})
+
+	jobs := []*plugin_pb.JobSpec{
+		replicaJob("a", "10.0.0.1:8080"),
+		replicaJob("b", "10.0.0.1:8080"),
+	}
+	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 0}
+	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
+
+	_, target := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if target != "worker-local" {
+		t.Fatalf("first take want local, got %s", target)
+	}
+	_, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local")
+	if err != nil {
+		t.Fatalf("reserve local: %v", err)
+	}
+	defer release()
+
+	stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if stolen == nil || stealTarget != "worker-remote" {
+		t.Fatalf("K=0 should steal immediately, got job=%v target=%s", stolen, stealTarget)
+	}
+}
+
+func TestBindQueuedJobWorkerKeepsPendingState(t *testing.T) {
+	t.Parallel()
+
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	job := replicaJob("bind-1", "10.0.0.1:8080")
+	pluginSvc.trackExecutionQueued(job)
+	pluginSvc.bindQueuedJobWorker(job, "worker-local")
+
+	pluginSvc.jobsMu.Lock()
+	tracked := pluginSvc.jobs[job.JobId]
+	pluginSvc.jobsMu.Unlock()
+	if tracked == nil {
+		t.Fatal("missing tracked job")
+	}
+	if tracked.State != "job_state_pending" {
+		t.Fatalf("state=%s", tracked.State)
+	}
+	if tracked.Stage != "queued" {
+		t.Fatalf("stage=%s", tracked.Stage)
+	}
+	if tracked.WorkerID != "worker-local" {
+		t.Fatalf("worker=%s", tracked.WorkerID)
+	}
+}