feat: wait before stealing a dead home worker queue

Orphaned queues are held for 30s so a worker reconnect blink does
not dump jobs onto other machines. After the grace, steal without K
as before. Zero disables the wait. A home that comes back during
grace keeps its queue.

weed/admin/plugin/plugin_scheduler.go

@@ -31,7 +31,10 @@ 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
+	defaultLocalQueuePin = 5
+	// Wait this long after a home worker disappears before stealing
+	// its queue; 0 steals immediately. Covers a reconnect blink.
+	defaultOrphanRehomeGrace         = 30 * time.Second
 	maxScheduledExecutionConcurrency = 128
 	defaultScheduledRetryBackoff     = 5 * time.Second
 	defaultClusterContextTimeout     = 10 * time.Second
@@ -56,6 +59,9 @@ type schedulerPolicy struct {
 	// LocalQueuePin is K: do not spill off replica-host workers while
 	// min(load(pref)) < K. Zero is the old immediate-spill path.
 	LocalQueuePin int
+	// OrphanRehomeGrace is how long a missing home's queue is held
+	// before steal-without-K. Zero steals as soon as the worker is gone.
+	OrphanRehomeGrace time.Duration
 }
 
 // laneSchedulerLoop is the main scheduling goroutine for a single lane.
@@ -474,6 +480,7 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err
 		RetryLimit:             int(adminRuntime.RetryLimit),
 		ExecutorReserveBackoff: 200 * time.Millisecond,
 		LocalQueuePin:          defaultLocalQueuePin,
+		OrphanRehomeGrace:      defaultOrphanRehomeGrace,
 	}
 
 	if policy.DetectionInterval < r.schedulerTick {
@@ -1650,6 +1657,7 @@ func (r *Plugin) DispatchProposals(
 			RetryBackoff:           defaultScheduledRetryBackoff,
 			ExecutorReserveBackoff: 200 * time.Millisecond,
 			LocalQueuePin:          defaultLocalQueuePin,
+			OrphanRehomeGrace:      defaultOrphanRehomeGrace,
 		}
 	}
 

weed/admin/plugin/plugin_scheduler_test.go

@@ -64,6 +64,9 @@ func TestLoadSchedulerPolicyUsesAdminConfig(t *testing.T) {
 	if policy.LocalQueuePin != defaultLocalQueuePin {
 		t.Fatalf("unexpected local queue pin: got=%d", policy.LocalQueuePin)
 	}
+	if policy.OrphanRehomeGrace != defaultOrphanRehomeGrace {
+		t.Fatalf("unexpected orphan rehome grace: got=%v", policy.OrphanRehomeGrace)
+	}
 }
 
 func TestLoadSchedulerPolicyUsesDescriptorDefaultsWhenConfigMissing(t *testing.T) {

weed/admin/plugin/worker_job_queues.go

@@ -2,6 +2,7 @@ package plugin
 
 import (
 	"sync"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 )
@@ -15,14 +16,18 @@ type queuedJob struct {
 // 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
+	mu           sync.Mutex
+	q            map[string][]*queuedJob
+	leftover     []*queuedJob
+	remaining    int
+	missingSince map[string]time.Time
 }
 
 func newWorkerJobQueues() *workerJobQueues {
-	return &workerJobQueues{q: make(map[string][]*queuedJob)}
+	return &workerJobQueues{
+		q:            make(map[string][]*queuedJob),
+		missingSince: make(map[string]time.Time),
+	}
 }
 
 func (qs *workerJobQueues) push(home string, job *plugin_pb.JobSpec) {
@@ -142,7 +147,8 @@ func (r *Plugin) assignJobsToWorkerQueues(
 
 // takeQueuedJob pops a home-runnable job, or steals onto a free worker.
 // Live homes are stolen from only when their queue is at/over K.
-// A home missing from live executors is stolen immediately (no K).
+// A home missing from live executors is stolen without K after
+// OrphanRehomeGrace (0 = immediately).
 func (r *Plugin) takeQueuedJob(
 	qs *workerJobQueues,
 	jobType string,
@@ -221,20 +227,33 @@ func (r *Plugin) takeQueuedJob(
 	live := make(map[string]struct{}, len(executors))
 	for _, exec := range executors {
 		live[exec.WorkerID] = struct{}{}
+		delete(qs.missingSince, exec.WorkerID)
 	}
 
+	now := time.Now()
+	grace := policy.OrphanRehomeGrace
 	victimID := ""
 	victimLen := 0
 	victimOrphan := false
 	for id, qq := range qs.q {
 		if len(qq) == 0 {
+			delete(qs.missingSince, id)
 			continue
 		}
 		_, isLive := live[id]
-		if isLive && pin > 0 && len(qq) < pin {
-			continue
-		}
 		orphan := !isLive
+		if isLive {
+			if pin > 0 && len(qq) < pin {
+				continue
+			}
+		} else {
+			if _, ok := qs.missingSince[id]; !ok {
+				qs.missingSince[id] = now
+			}
+			if grace > 0 && now.Sub(qs.missingSince[id]) < grace {
+				continue
+			}
+		}
 		better := victimID == ""
 		if !better && orphan != victimOrphan {
 			better = orphan

weed/admin/plugin/worker_job_queues_test.go

@@ -3,6 +3,7 @@ package plugin
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 )
@@ -218,7 +219,11 @@ func TestTakeStealsOrphanedHomeWithoutK(t *testing.T) {
 		replicaJob("o1", "10.0.0.1:8080"),
 		replicaJob("o2", "10.0.0.1:8080"),
 	}
-	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5}
+	policy := schedulerPolicy{
+		PerWorkerConcurrency: 1,
+		LocalQueuePin:        5,
+		OrphanRehomeGrace:    30 * time.Second,
+	}
 	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
 	if qs.queueLen("worker-local") != 2 {
 		t.Fatalf("want 2 on local, got %d", qs.queueLen("worker-local"))
@@ -226,9 +231,20 @@ func TestTakeStealsOrphanedHomeWithoutK(t *testing.T) {
 
 	pluginSvc.registry.Remove("worker-local")
 
+	if item, tgt := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy); item != nil {
+		t.Fatalf("within grace should hold orphan queue, got target=%s", tgt)
+	}
+	if qs.queueLen("worker-local") != 2 {
+		t.Fatalf("held queue changed: %d", qs.queueLen("worker-local"))
+	}
+
+	qs.mu.Lock()
+	qs.missingSince["worker-local"] = time.Now().Add(-time.Minute)
+	qs.mu.Unlock()
+
 	stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
 	if stolen == nil || stealTarget != "worker-remote" {
-		t.Fatalf("dead home should steal without K, got job=%v target=%s", stolen, stealTarget)
+		t.Fatalf("after grace should steal without K, got job=%v target=%s", stolen, stealTarget)
 	}
 	if qs.queueLen("worker-local") != 1 {
 		t.Fatalf("want 1 orphan left, got %d", qs.queueLen("worker-local"))
@@ -240,6 +256,64 @@ func TestTakeStealsOrphanedHomeWithoutK(t *testing.T) {
 	}
 }
 
+func TestTakeKeepsOrphanWhenHomeReconnectsDuringGrace(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("blink", "10.0.0.1:8080")}
+	policy := schedulerPolicy{
+		PerWorkerConcurrency: 1,
+		LocalQueuePin:        5,
+		OrphanRehomeGrace:    30 * time.Second,
+	}
+	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
+	pluginSvc.registry.Remove("worker-local")
+	if item, _ := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy); item != nil {
+		t.Fatal("expected hold during grace")
+	}
+
+	registerECWorkers(t, pluginSvc, map[string]string{"worker-local": "10.0.0.1:9334"})
+	item, target := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if item == nil || target != "worker-local" {
+		t.Fatalf("reconnected home should run its queue, got job=%v target=%s", item, target)
+	}
+}
+
+func TestTakeStealsOrphanImmediatelyWhenGraceDisabled(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("o1", "10.0.0.1:8080")}
+	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5, OrphanRehomeGrace: 0}
+	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
+	pluginSvc.registry.Remove("worker-local")
+
+	stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if stolen == nil || stealTarget != "worker-remote" {
+		t.Fatalf("grace=0 should steal immediately, got job=%v target=%s", stolen, stealTarget)
+	}
+}
+
 func TestBindQueuedJobWorkerKeepsPendingState(t *testing.T) {
 	t.Parallel()