fix: rehome queued jobs when the home worker dies

A live home still pins below K. A home missing from live executors
is stolen immediately onto a free worker so a short orphaned queue
does not sit until the dispatch window ends.

weed/admin/plugin/worker_job_queues.go

@@ -140,8 +140,9 @@ func (r *Plugin) assignJobsToWorkerQueues(
 	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.
+// 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).
 func (r *Plugin) takeQueuedJob(
 	qs *workerJobQueues,
 	jobType string,
@@ -156,6 +157,11 @@ func (r *Plugin) takeQueuedJob(
 	}
 	lane := JobTypeLane(jobType)
 	pin := policy.LocalQueuePin
+	defer func() {
+		if item != nil && targetID != "" {
+			r.bindQueuedJobWorker(item.job, targetID)
+		}
+	}()
 
 	qs.mu.Lock()
 	defer qs.mu.Unlock()
@@ -212,18 +218,34 @@ func (r *Plugin) takeQueuedJob(
 		return item, thief.WorkerID
 	}
 
+	live := make(map[string]struct{}, len(executors))
+	for _, exec := range executors {
+		live[exec.WorkerID] = struct{}{}
+	}
+
 	victimID := ""
 	victimLen := 0
+	victimOrphan := false
 	for id, qq := range qs.q {
 		if len(qq) == 0 {
 			continue
 		}
-		if pin > 0 && len(qq) < pin {
+		_, isLive := live[id]
+		if isLive && pin > 0 && len(qq) < pin {
 			continue
 		}
-		if victimID == "" || len(qq) > victimLen || (len(qq) == victimLen && id < victimID) {
+		orphan := !isLive
+		better := victimID == ""
+		if !better && orphan != victimOrphan {
+			better = orphan
+		}
+		if !better && orphan == victimOrphan {
+			better = len(qq) > victimLen || (len(qq) == victimLen && id < victimID)
+		}
+		if better {
 			victimID = id
 			victimLen = len(qq)
+			victimOrphan = orphan
 		}
 	}
 	if victimID == "" {

weed/admin/plugin/worker_job_queues_test.go

@@ -200,6 +200,46 @@ func TestTakeStealsImmediatelyWhenPinDisabled(t *testing.T) {
 	}
 }
 
+func TestTakeStealsOrphanedHomeWithoutK(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"),
+		replicaJob("o2", "10.0.0.1:8080"),
+	}
+	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5}
+	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"))
+	}
+
+	pluginSvc.registry.Remove("worker-local")
+
+	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)
+	}
+	if qs.queueLen("worker-local") != 1 {
+		t.Fatalf("want 1 orphan left, got %d", qs.queueLen("worker-local"))
+	}
+
+	stolen2, stealTarget2 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
+	if stolen2 == nil || stealTarget2 != "worker-remote" {
+		t.Fatalf("second orphan should also move, got job=%v target=%s", stolen2, stealTarget2)
+	}
+}
+
 func TestBindQueuedJobWorkerKeepsPendingState(t *testing.T) {
 	t.Parallel()