feat: prefer same-host worker and local EC replica

Admin executor pick ranked idle workers only, so a job whose volume
lived on full-03 could still run on full-01 and pull the whole .dat
across the LAN. Rank workers whose Hello host matches any replica
first (fallback to others if the local worker is full).

Detection named Sources[0] / the lowest server id; execute then
SyncAndSelectBestReplica and could switch the copy source to another
rack. After sync, if a replica on this machine has a complete file
count, copy from that URL so affinity and pull stay on one host.

weed/admin/plugin/host_affinity.go

@@ -0,0 +1,66 @@
+package plugin
+
+import (
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+
+	"google.golang.org/protobuf/proto"
+)
+
+// JobReplicaHosts are the volume-replica hosts a job will read from.
+// Used so executor pick prefers a worker on one of those machines.
+// Sources with ShardIds are EC-cleanup rows, not encode replicas.
+func JobReplicaHosts(job *plugin_pb.JobSpec) []string {
+	if job == nil {
+		return nil
+	}
+	var addrs []string
+	addrs = append(addrs, pluginworker.ReadStringListConfig(job.Parameters, "replica_servers")...)
+	if src := strings.TrimSpace(pluginworker.ReadStringConfig(job.Parameters, "source_server", "")); src != "" {
+		addrs = append(addrs, src)
+	}
+	if payload := pluginworker.ReadBytesConfig(job.Parameters, "task_params_pb"); len(payload) > 0 {
+		params := &worker_pb.TaskParams{}
+		if err := proto.Unmarshal(payload, params); err == nil {
+			for _, s := range params.GetSources() {
+				if s == nil || s.Node == "" || len(s.ShardIds) > 0 {
+					continue
+				}
+				addrs = append(addrs, s.Node)
+			}
+		}
+	}
+	return util.HostKeysFromAddresses(addrs)
+}
+
+// partitionByPreferredHost splits candidates into (host matches, others).
+// No preferred hosts, or no matches, returns (nil, all) so callers keep
+// the existing slots+round-robin path.
+func partitionByPreferredHost(candidates []*WorkerSession, preferredHosts []string) (pref, rest []*WorkerSession) {
+	want := util.HostKeysFromAddresses(preferredHosts)
+	if len(want) == 0 || len(candidates) == 0 {
+		return nil, candidates
+	}
+	set := make(map[string]struct{}, len(want))
+	for _, h := range want {
+		set[h] = struct{}{}
+	}
+	for _, c := range candidates {
+		if c == nil {
+			continue
+		}
+		if _, ok := set[util.HostKey(c.Address)]; ok {
+			pref = append(pref, c)
+			continue
+		}
+		rest = append(rest, c)
+	}
+	if len(pref) == 0 {
+		return nil, candidates
+	}
+	return pref, rest
+}

weed/admin/plugin/plugin.go

@@ -609,7 +609,7 @@ func (r *Plugin) ExecuteJob(
 	releaseGate := r.acquireDetectionExecutionGate(jobType, true)
 	defer releaseGate()
 
-	executor, err := r.registry.PickExecutor(jobType)
+	executor, err := r.registry.PickExecutor(jobType, JobReplicaHosts(job)...)
 	if err != nil {
 		return nil, err
 	}

weed/admin/plugin/plugin_scheduler.go

@@ -873,7 +873,7 @@ func (r *Plugin) dispatchScheduledProposals(
 						continue jobLoop
 					}
 
-					executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy)
+					executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy, JobReplicaHosts(job))
 					if reserveErr != nil {
 						if ctx.Err() != nil {
 							r.cancelQueuedJob(job, ctx.Err())
@@ -988,6 +988,7 @@ func (r *Plugin) reserveScheduledExecutor(
 	ctx context.Context,
 	jobType string,
 	policy schedulerPolicy,
+	preferredHosts []string,
 ) (*WorkerSession, func(), error) {
 	if ctx == nil {
 		ctx = context.Background()
@@ -1015,7 +1016,7 @@ func (r *Plugin) reserveScheduledExecutor(
 			return nil, nil, fmt.Errorf("timed out waiting for executor capacity for %s", jobType)
 		}
 
-		executors, err := r.registry.ListExecutors(jobType)
+		executors, err := r.registry.ListExecutors(jobType, preferredHosts...)
 		if err != nil {
 			if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
 				if ctx.Err() != nil {

weed/admin/plugin/plugin_scheduler_test.go

@@ -138,13 +138,13 @@ func TestReserveScheduledExecutorRespectsPerWorkerLimit(t *testing.T) {
 		ExecutorReserveBackoff: time.Millisecond,
 	}
 
-	executor1, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
+	executor1, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
 	if err != nil {
 		t.Fatalf("reserve executor 1: %v", err)
 	}
 	defer release1()
 
-	executor2, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
+	executor2, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
 	if err != nil {
 		t.Fatalf("reserve executor 2: %v", err)
 	}
@@ -266,7 +266,7 @@ func TestReserveScheduledExecutorTimesOutWhenNoExecutor(t *testing.T) {
 
 	start := time.Now()
 	pluginSvc.Shutdown()
-	_, _, err = pluginSvc.reserveScheduledExecutor(context.Background(), "missing-job-type", policy)
+	_, _, err = pluginSvc.reserveScheduledExecutor(context.Background(), "missing-job-type", policy, nil)
 	if err == nil {
 		t.Fatalf("expected reservation shutdown error")
 	}
@@ -297,7 +297,7 @@ func TestReserveScheduledExecutorWaitsForWorkerCapacity(t *testing.T) {
 		ExecutorReserveBackoff: 5 * time.Millisecond,
 	}
 
-	_, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
+	_, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
 	if err != nil {
 		t.Fatalf("reserve executor 1: %v", err)
 	}
@@ -308,7 +308,7 @@ func TestReserveScheduledExecutorWaitsForWorkerCapacity(t *testing.T) {
 	}
 	secondReserveCh := make(chan reserveResult, 1)
 	go func() {
-		_, release2, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
+		_, release2, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
 		if release2 != nil {
 			release2()
 		}

weed/admin/plugin/registry.go

@@ -277,14 +277,24 @@ func (r *Registry) PickDetector(jobType string) (*WorkerSession, error) {
 }
 
 // PickExecutor picks one executor worker for a job type.
-func (r *Registry) PickExecutor(jobType string) (*WorkerSession, error) {
-	return r.pickByKind(jobType, false)
+// preferredHosts, when set, ranks workers whose Hello Address host matches a
+// replica/source host ahead of others (same-machine EC pull). Empty means the
+// previous slots+round-robin order.
+func (r *Registry) PickExecutor(jobType string, preferredHosts ...string) (*WorkerSession, error) {
+	list, err := r.ListExecutors(jobType, preferredHosts...)
+	if err != nil {
+		return nil, err
+	}
+	return list[0], nil
 }
 
 // ListExecutors returns sorted executor candidates for one job type.
 // Ordering is by most available execution slots, then lexical worker ID.
 // The top tie group is rotated round-robin to prevent sticky assignment.
-func (r *Registry) ListExecutors(jobType string) ([]*WorkerSession, error) {
+// preferredHosts, when any worker matches, are listed first (then the rest);
+// rotation applies only inside the preferred group so a local worker is not
+// rotated away in favor of a remote one.
+func (r *Registry) ListExecutors(jobType string, preferredHosts ...string) ([]*WorkerSession, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -293,8 +303,16 @@ func (r *Registry) ListExecutors(jobType string) ([]*WorkerSession, error) {
 		return nil, fmt.Errorf("no executor worker available for job_type=%s", jobType)
 	}
 
-	sortByKind(candidates, jobType, false)
-	r.rotateTopCandidatesLocked(candidates, jobType, false)
+	pref, rest := partitionByPreferredHost(candidates, preferredHosts)
+	if len(pref) > 0 {
+		sortByKind(pref, jobType, false)
+		r.rotateTopCandidatesLocked(pref, jobType, false)
+		sortByKind(rest, jobType, false)
+		candidates = append(pref, rest...)
+	} else {
+		sortByKind(candidates, jobType, false)
+		r.rotateTopCandidatesLocked(candidates, jobType, false)
+	}
 
 	out := make([]*WorkerSession, 0, len(candidates))
 	for _, candidate := range candidates {

weed/admin/plugin/registry_test.go

@@ -401,3 +401,70 @@ func TestRegistryHasCapableWorkerIgnoresNilCapability(t *testing.T) {
 		t.Fatalf("expected HasCapableWorker to return false when capability has no detect or execute")
 	}
 }
+
+func TestListExecutorsPrefersMatchingReplicaHost(t *testing.T) {
+	t.Parallel()
+
+	r := NewRegistry()
+	r.UpsertFromHello(&plugin_pb.WorkerHello{
+		WorkerId: "worker-remote",
+		Address:  "10.87.0.4:33646",
+		Capabilities: []*plugin_pb.JobTypeCapability{
+			{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 8},
+		},
+	})
+	r.UpsertFromHello(&plugin_pb.WorkerHello{
+		WorkerId: "worker-local",
+		Address:  "10.87.0.3:33646",
+		Capabilities: []*plugin_pb.JobTypeCapability{
+			{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 1},
+		},
+	})
+	r.UpdateHeartbeat("worker-remote", &plugin_pb.WorkerHeartbeat{
+		WorkerId: "worker-remote", ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 8,
+	})
+	r.UpdateHeartbeat("worker-local", &plugin_pb.WorkerHeartbeat{
+		WorkerId: "worker-local", ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 1,
+	})
+
+	// Without preference the emptier remote worker wins on slots.
+	plain, err := r.ListExecutors("erasure_coding")
+	if err != nil {
+		t.Fatalf("ListExecutors: %v", err)
+	}
+	if plain[0].WorkerID != "worker-remote" {
+		t.Fatalf("without preference want remote first, got %s", plain[0].WorkerID)
+	}
+
+	// Replica on :8080 of the same host as worker-local.
+	pref, err := r.ListExecutors("erasure_coding", "10.87.0.3:8080")
+	if err != nil {
+		t.Fatalf("ListExecutors preferred: %v", err)
+	}
+	if pref[0].WorkerID != "worker-local" {
+		t.Fatalf("preferred host want worker-local first, got %s", pref[0].WorkerID)
+	}
+	if pref[1].WorkerID != "worker-remote" {
+		t.Fatalf("remote should still be fallback, got %s", pref[1].WorkerID)
+	}
+}
+
+func TestJobReplicaHostsFromSourcesAndReplicaServers(t *testing.T) {
+	t.Parallel()
+	job := &plugin_pb.JobSpec{
+		Parameters: map[string]*plugin_pb.ConfigValue{
+			"source_server": {
+				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "10.87.0.1:8080"},
+			},
+			"replica_servers": {
+				Kind: &plugin_pb.ConfigValue_StringList{
+					StringList: &plugin_pb.StringList{Values: []string{"10.87.0.1:8080", "10.87.0.2:8081"}},
+				},
+			},
+		},
+	}
+	got := JobReplicaHosts(job)
+	if len(got) != 2 {
+		t.Fatalf("got %v", got)
+	}
+}

weed/util/net_host.go

@@ -0,0 +1,79 @@
+package util
+
+import (
+	"net"
+	"strings"
+)
+
+// HostKey is the hostname/IP of an "host:port" address (or the string itself
+// when it has no port). Compared case-insensitively so worker Hello
+// "10.87.0.3:33646" matches a volume replica on "10.87.0.3:8080".
+func HostKey(addr string) string {
+	addr = strings.TrimSpace(addr)
+	if addr == "" {
+		return ""
+	}
+	host, _, err := net.SplitHostPort(addr)
+	if err != nil {
+		return strings.ToLower(addr)
+	}
+	return strings.ToLower(host)
+}
+
+// HostKeysFromAddresses returns unique HostKey values, skipping empties.
+func HostKeysFromAddresses(addrs []string) []string {
+	seen := make(map[string]struct{}, len(addrs))
+	var out []string
+	for _, a := range addrs {
+		h := HostKey(a)
+		if h == "" {
+			continue
+		}
+		if _, ok := seen[h]; ok {
+			continue
+		}
+		seen[h] = struct{}{}
+		out = append(out, h)
+	}
+	return out
+}
+
+// AddressIsLocal reports whether addr's host is this process's machine
+// (loopback or a configured interface address). Used so an EC worker prefers
+// a replica on the same box after replica sync.
+func AddressIsLocal(addr string) bool {
+	host := HostKey(addr)
+	if host == "" {
+		return false
+	}
+	if host == "localhost" || host == "127.0.0.1" || host == "::1" {
+		return true
+	}
+	ip := net.ParseIP(host)
+	if ip != nil && ip.IsLoopback() {
+		return true
+	}
+	ifaces, err := net.InterfaceAddrs()
+	if err != nil {
+		return false
+	}
+	for _, ia := range ifaces {
+		var got net.IP
+		switch v := ia.(type) {
+		case *net.IPNet:
+			got = v.IP
+		case *net.IPAddr:
+			got = v.IP
+		}
+		if got == nil {
+			continue
+		}
+		if strings.EqualFold(got.String(), host) {
+			return true
+		}
+		if ip != nil && got.Equal(ip) {
+			return true
+		}
+	}
+	return false
+}

weed/util/net_host_test.go

@@ -0,0 +1,24 @@
+package util
+
+import "testing"
+
+func TestHostKeyStripsPort(t *testing.T) {
+	t.Parallel()
+	if got := HostKey("10.87.0.3:8080"); got != "10.87.0.3" {
+		t.Fatalf("HostKey replica: got %q", got)
+	}
+	if got := HostKey("10.87.0.3:33646"); got != "10.87.0.3" {
+		t.Fatalf("HostKey worker: got %q", got)
+	}
+	if HostKey("10.87.0.3:8080") != HostKey("10.87.0.3:33646") {
+		t.Fatal("worker and volume server on the same IP must share a HostKey")
+	}
+}
+
+func TestHostKeysFromAddressesDedups(t *testing.T) {
+	t.Parallel()
+	got := HostKeysFromAddresses([]string{"10.0.0.1:8080", "10.0.0.1:8081", "10.0.0.2:8080", ""})
+	if len(got) != 2 || got[0] != "10.0.0.1" || got[1] != "10.0.0.2" {
+		t.Fatalf("got %v", got)
+	}
+}

weed/worker/tasks/erasure_coding/ec_task.go

@@ -21,6 +21,7 @@ import (
 	storagetypes "github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
 	"github.com/seaweedfs/seaweedfs/weed/storage/volume_replica"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types/base"
@@ -403,9 +404,52 @@ func (t *ErasureCodingTask) syncAndSelectSourceReplica() error {
 		glog.Infof("EC encode: using best replica %s as source for volume %d (was %s)", best.Url, t.volumeID, t.server)
 		t.server = best.Url
 	}
+	// Host-affinity: if this worker sits on a replica and that copy is as
+	// complete as the union (typical 010 after sync), pull from localhost
+	// instead of the lexically-first / largest remote replica.
+	if local := preferLocalReplicaIfComplete(t.grpcDialOption, needle.VolumeId(t.volumeID), locs, t.server); local != "" && local != t.server {
+		glog.Infof("EC encode: preferring local replica %s as source for volume %d (was %s)", local, t.volumeID, t.server)
+		t.server = local
+	}
 	return nil
 }
 
+// preferLocalReplicaIfComplete returns a replica URL on this machine whose
+// file count is >= the best remote replica. After SyncAndSelectBestReplica
+// a consistent set is equal everywhere; an inconsistent union lives only on
+// the chosen best, so a stale local copy is not selected.
+func preferLocalReplicaIfComplete(dial grpc.DialOption, vid needle.VolumeId, locs []wdclient.Location, current string) string {
+	var localURL string
+	for _, loc := range locs {
+		if util.AddressIsLocal(loc.Url) {
+			localURL = loc.Url
+			break
+		}
+	}
+	if localURL == "" || localURL == current {
+		return current
+	}
+	statuses := volume_replica.GetReplicaStatuses(dial, vid, locs)
+	var localCount, maxCount uint64
+	var sawLocal bool
+	for _, s := range statuses {
+		if s.Error != nil {
+			continue
+		}
+		if s.FileCount > maxCount {
+			maxCount = s.FileCount
+		}
+		if s.Location.Url == localURL {
+			localCount = s.FileCount
+			sawLocal = true
+		}
+	}
+	if !sawLocal || localCount < maxCount {
+		return current
+	}
+	return localURL
+}
+
 // rollbackReadonly is a best-effort restore of every replica markReplicasReadonly
 // touched, used when the EC task fails before the originals are deleted. Logs but
 // does not return errors; uses a fresh context since the caller's may be cancelled.

weed/worker/tasks/erasure_coding/plugin_handler.go

@@ -670,8 +670,15 @@ func buildErasureCodingProposal(
 	}
 
 	sourceNode := ""
-	if len(params.Sources) > 0 {
-		sourceNode = strings.TrimSpace(params.Sources[0].Node)
+	var replicaServers []string
+	for _, s := range params.Sources {
+		if s == nil || strings.TrimSpace(s.Node) == "" || len(s.ShardIds) > 0 {
+			continue
+		}
+		replicaServers = append(replicaServers, s.Node)
+		if sourceNode == "" {
+			sourceNode = strings.TrimSpace(s.Node)
+		}
 	}
 
 	summary := fmt.Sprintf("Erasure code volume %d", result.VolumeID)
@@ -702,6 +709,9 @@ func buildErasureCodingProposal(
 			"source_server": {
 				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: sourceNode},
 			},
+			"replica_servers": {
+				Kind: &plugin_pb.ConfigValue_StringList{StringList: &plugin_pb.StringList{Values: replicaServers}},
+			},
 			"collection": {
 				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: result.Collection},
 			},