Merge remote-tracking branch 'origin/master' into pr11709-merge

# Conflicts:
#	weed/command/filer_remote_sync_dir.go
#	weed/command/filer_remote_sync_dir_test.go

.github/workflows/codeql.yml

@@ -27,7 +27,7 @@ jobs:
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@v4.38.3
+        uses: github/codeql-action/init@v4.38.2
         # Override language selection by uncommenting this and choosing your languages
         with:
           languages: go
@@ -35,7 +35,7 @@ jobs:
       # Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
       # If this step fails, then you should remove it and run the build manually (see below).
       - name: Autobuild
-        uses: github/codeql-action/autobuild@v4.38.3
+        uses: github/codeql-action/autobuild@v4.38.2
 
       # ℹ️ Command-line programs to run using the OS shell.
       # 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
@@ -49,4 +49,4 @@ jobs:
       #     make release
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@v4.38.3
+        uses: github/codeql-action/analyze@v4.38.2

.github/workflows/container_latest.yml

@@ -405,7 +405,7 @@ jobs:
           output: trivy-results.sarif
           exit-code: '0'
       - name: Upload Trivy scan results to GitHub Security
-        uses: github/codeql-action/upload-sarif@v4.38.3
+        uses: github/codeql-action/upload-sarif@v4.38.2
         if: always()
         with:
           sarif_file: trivy-results.sarif

.github/workflows/container_release_unified.yml

@@ -460,7 +460,7 @@ jobs:
 
       - name: Upload Trivy scan results to GitHub Security
         if: always()
-        uses: github/codeql-action/upload-sarif@v4.38.3
+        uses: github/codeql-action/upload-sarif@v4.38.2
         with:
           sarif_file: trivy-results.sarif
           category: trivy-${{ matrix.variant }}

seaweed-volume/Cargo.lock

@@ -234,9 +234,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.1"
+version = "1.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e"
+checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc"
 dependencies = [
  "aws-lc-sys",
  "zeroize",
@@ -244,15 +244,14 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.45.0"
+version = "0.39.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27"
+checksum = "83a25cf98105baa966497416dbd42565ce3a8cf8dbfd59803ec9ad46f3126399"
 dependencies = [
  "cc",
  "cmake",
  "dunce",
  "fs_extra",
- "pkg-config",
 ]
 
 [[package]]
@@ -2678,12 +2677,6 @@ dependencies = [
  "spki 0.7.3",
 ]
 
-[[package]]
-name = "pkg-config"
-version = "0.3.34"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548"
-
 [[package]]
 name = "png"
 version = "0.18.1"
@@ -3396,9 +3389,9 @@ dependencies = [
 
 [[package]]
 name = "rustls"
-version = "0.23.45"
+version = "0.23.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634"
+checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4"
 dependencies = [
  "aws-lc-rs",
  "log",
@@ -3443,9 +3436,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-webpki"
-version = "0.103.15"
+version = "0.103.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2"
+checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
 dependencies = [
  "aws-lc-rs",
  "ring",

test/sftp/basic_test.go

@@ -3,7 +3,6 @@ package sftp
 import (
 	"bytes"
 	"io"
-	"os"
 	"path"
 	"testing"
 
@@ -603,32 +602,6 @@ func TestFileContent(t *testing.T) {
 		sftpClient.Remove("/binary.bin")
 	})
 
-	t.Run("WriteAtOffsetKeepsExistingBytes", func(t *testing.T) {
-		file, err := sftpClient.Create("/resume.txt")
-		require.NoError(t, err)
-		_, err = file.Write([]byte("hello world"))
-		require.NoError(t, err)
-		require.NoError(t, file.Close())
-
-		// Reopen without O_TRUNC and write past the start, as a resumed upload does
-		file, err = sftpClient.OpenFile("/resume.txt", os.O_WRONLY)
-		require.NoError(t, err)
-		_, err = file.WriteAt([]byte("WORLD!"), 6)
-		require.NoError(t, err)
-		require.NoError(t, file.Close())
-
-		readFile, err := sftpClient.Open("/resume.txt")
-		require.NoError(t, err)
-		content, err := io.ReadAll(readFile)
-		require.NoError(t, err)
-		readFile.Close()
-
-		require.Equal(t, "hello WORLD!", string(content))
-
-		// Clean up
-		sftpClient.Remove("/resume.txt")
-	})
-
 	t.Run("EmptyFile", func(t *testing.T) {
 		file, err := sftpClient.Create("/empty.txt")
 		require.NoError(t, err)

weed/admin/plugin/host_affinity.go

@@ -1,66 +0,0 @@
-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

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

weed/admin/plugin/plugin_monitor.go

@@ -887,37 +887,6 @@ 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

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -29,18 +28,12 @@ const (
 	defaultScheduledMaxResults           int32 = 1000
 	defaultScheduledExecutionConcurrency       = 1
 	defaultScheduledPerWorkerConcurrency       = 1
-	// Hold a job on replica-host workers until their reservation load
-	// reaches this; 0 disables the pin and spills immediately.
-	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
-	defaultWaitingBacklogFloor       = 8
-	defaultWaitingBacklogMultiplier  = 4
-	maxEstimatedRuntimeCap           = 8 * time.Hour
+	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
 )
@@ -56,12 +49,6 @@ type schedulerPolicy struct {
 	PerWorkerConcurrency   int
 	RetryLimit             int
 	ExecutorReserveBackoff time.Duration
-	// 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.
@@ -479,8 +466,6 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err
 		PerWorkerConcurrency:   int(adminRuntime.PerWorkerExecutionConcurrency),
 		RetryLimit:             int(adminRuntime.RetryLimit),
 		ExecutorReserveBackoff: 200 * time.Millisecond,
-		LocalQueuePin:          defaultLocalQueuePin,
-		OrphanRehomeGrace:      defaultOrphanRehomeGrace,
 	}
 
 	if policy.DetectionInterval < r.schedulerTick {
@@ -511,9 +496,6 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err
 	if policy.ExecutionTimeout < defaultScheduledExecutionTimeout {
 		policy.ExecutionTimeout = defaultScheduledExecutionTimeout
 	}
-	if policy.LocalQueuePin < 0 {
-		policy.LocalQueuePin = 0
-	}
 
 	return policy, true, nil
 }
@@ -831,22 +813,19 @@ func (r *Plugin) dispatchScheduledProposals(
 	// whole dispatch lets manual shell operations interleave between jobs.
 	jobsRequireLock := LaneRequiresLock(JobTypeLane(jobType))
 
-	jobs := make([]*plugin_pb.JobSpec, 0, len(proposals))
+	jobQueue := make(chan *plugin_pb.JobSpec, len(proposals))
 	for index, proposal := range proposals {
-		select {
-		case <-r.shutdownCh:
-			for _, job := range jobs {
-				r.cancelQueuedJob(job, errSchedulerShutdown)
-			}
-			return 0, 0, len(jobs)
-		default:
-		}
 		job := buildScheduledJobSpec(jobType, proposal, index)
 		r.trackExecutionQueued(job)
-		jobs = append(jobs, job)
+		select {
+		case <-r.shutdownCh:
+			close(jobQueue)
+			return 0, 0, 0
+		default:
+			jobQueue <- job
+		}
 	}
-
-	queues := r.assignJobsToWorkerQueues(jobType, jobs, policy)
+	close(jobQueue)
 
 	var wg sync.WaitGroup
 	var statsMu sync.Mutex
@@ -863,110 +842,130 @@ func (r *Plugin) dispatchScheduledProposals(
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			for {
+
+		jobLoop:
+			for job := range jobQueue {
 				select {
 				case <-r.shutdownCh:
 					return
 				default:
 				}
-				if ctx.Err() != nil {
-					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
-					}
+				if ctx.Err() != nil {
+					r.cancelQueuedJob(job, ctx.Err())
+					statsMu.Lock()
+					canceledCount++
+					statsMu.Unlock()
 					continue
 				}
-				if ctx.Err() != nil {
-					queues.pushFront(item)
-					return
-				}
 
-				executor, release, reserveErr := r.reserveExecutorByID(jobType, policy, targetID)
-				if reserveErr != nil {
-					queues.pushFront(item)
+				for {
+					select {
+					case <-r.shutdownCh:
+						return
+					default:
+					}
 					if ctx.Err() != nil {
-						return
+						r.cancelQueuedJob(job, ctx.Err())
+						statsMu.Lock()
+						canceledCount++
+						statsMu.Unlock()
+						continue jobLoop
 					}
-					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
-						return
-					}
-					continue
-				}
 
-				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()
+					executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy)
+					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{
-							JobID:      item.job.JobId,
-							JobType:    item.job.JobType,
+							JobType:    jobType,
 							Source:     "admin_scheduler",
-							Message:    fmt.Sprintf("scheduled execution lock failed: %v", lockErr),
+							Message:    fmt.Sprintf("scheduled execution reservation failed: %v", reserveErr),
 							Stage:      "failed",
 							OccurredAt: timeToPtr(time.Now().UTC()),
 						})
-						continue
+						break
 					}
-					if ctx.Err() != nil {
+
+					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()
-						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()
+					release()
+					if errors.Is(err, errExecutorAtCapacity) {
+						r.trackExecutionQueued(job)
+						if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
+							return
+						}
 						continue
 					}
-					statsMu.Lock()
-					errorCount++
-					statsMu.Unlock()
-					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
-				}
+					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++
-				statsMu.Unlock()
+					statsMu.Lock()
+					successCount++
+					statsMu.Unlock()
+					break
+				}
 			}
 		}()
 	}
@@ -977,8 +976,8 @@ func (r *Plugin) dispatchScheduledProposals(
 	if drainErr == nil {
 		drainErr = errSchedulerShutdown
 	}
-	for _, item := range queues.drain() {
-		r.cancelQueuedJob(item.job, drainErr)
+	for job := range jobQueue {
+		r.cancelQueuedJob(job, drainErr)
 		canceledCount++
 	}
 
@@ -989,7 +988,6 @@ func (r *Plugin) reserveScheduledExecutor(
 	ctx context.Context,
 	jobType string,
 	policy schedulerPolicy,
-	preferredHosts []string,
 ) (*WorkerSession, func(), error) {
 	if ctx == nil {
 		ctx = context.Background()
@@ -1017,7 +1015,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, preferredHosts...)
+		executors, err := r.registry.ListExecutors(jobType)
 		if err != nil {
 			if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
 				if ctx.Err() != nil {
@@ -1028,19 +1026,11 @@ func (r *Plugin) reserveScheduledExecutor(
 			continue
 		}
 
-		pref, rest := partitionByPreferredHost(executors, preferredHosts)
-		lane := JobTypeLane(jobType)
-		loadOf := func(e *WorkerSession) int { return r.executorReservationLoad(e, lane) }
-
-		if len(pref) > 0 {
-			sortExecutorsByLoad(pref, loadOf)
-			if executor, release, ok := r.tryReserveFirst(pref, jobType, policy); ok {
-				return executor, release, nil
+		for _, executor := range executors {
+			release, ok := r.tryReserveExecutorCapacity(executor, jobType, policy)
+			if !ok {
+				continue
 			}
-		}
-
-		sortExecutorsByLoad(rest, loadOf)
-		if executor, release, ok := r.tryReserveFirst(rest, jobType, policy); ok {
 			return executor, release, nil
 		}
 
@@ -1061,75 +1051,6 @@ func (r *Plugin) tryReserveExecutorCapacity(
 	return r.tryReserveExecutorCapacityForLane(executor, jobType, policy, JobTypeLane(jobType))
 }
 
-func (r *Plugin) tryReserveFirst(executors []*WorkerSession, jobType string, policy schedulerPolicy) (*WorkerSession, func(), bool) {
-	for _, executor := range executors {
-		release, ok := r.tryReserveExecutorCapacity(executor, jobType, policy)
-		if ok {
-			return executor, release, true
-		}
-	}
-	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
-	}
-	load := 0
-	if executor.Heartbeat != nil && executor.Heartbeat.ExecutionSlotsUsed > 0 {
-		load = int(executor.Heartbeat.ExecutionSlotsUsed)
-	}
-	workerID := strings.TrimSpace(executor.WorkerID)
-	if ls := r.lanes[lane]; ls != nil {
-		ls.execMu.Lock()
-		load += ls.execRes[workerID]
-		ls.execMu.Unlock()
-		return load
-	}
-	r.schedulerExecMu.Lock()
-	load += r.schedulerExecReservations[workerID]
-	r.schedulerExecMu.Unlock()
-	return load
-}
-
-func sortExecutorsByLoad(executors []*WorkerSession, load func(*WorkerSession) int) {
-	sort.SliceStable(executors, func(i, j int) bool {
-		if executors[i] == nil {
-			return false
-		}
-		if executors[j] == nil {
-			return true
-		}
-		li, lj := load(executors[i]), load(executors[j])
-		if li != lj {
-			return li < lj
-		}
-		return executors[i].WorkerID < executors[j].WorkerID
-	})
-}
-
 // tryReserveExecutorCapacityForLane reserves an execution slot on the
 // per-lane reservation pool so that lanes cannot starve each other.
 func (r *Plugin) tryReserveExecutorCapacityForLane(
@@ -1656,8 +1577,6 @@ func (r *Plugin) DispatchProposals(
 			ExecutionTimeout:       defaultScheduledExecutionTimeout,
 			RetryBackoff:           defaultScheduledRetryBackoff,
 			ExecutorReserveBackoff: 200 * time.Millisecond,
-			LocalQueuePin:          defaultLocalQueuePin,
-			OrphanRehomeGrace:      defaultOrphanRehomeGrace,
 		}
 	}
 

weed/admin/plugin/plugin_scheduler_test.go

@@ -61,12 +61,6 @@ func TestLoadSchedulerPolicyUsesAdminConfig(t *testing.T) {
 	if policy.JobTypeMaxRuntime != 30*time.Minute {
 		t.Fatalf("unexpected max runtime: got=%v", policy.JobTypeMaxRuntime)
 	}
-	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) {
@@ -144,13 +138,13 @@ func TestReserveScheduledExecutorRespectsPerWorkerLimit(t *testing.T) {
 		ExecutorReserveBackoff: time.Millisecond,
 	}
 
-	executor1, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
+	executor1, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
 	if err != nil {
 		t.Fatalf("reserve executor 1: %v", err)
 	}
 	defer release1()
 
-	executor2, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
+	executor2, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
 	if err != nil {
 		t.Fatalf("reserve executor 2: %v", err)
 	}
@@ -272,7 +266,7 @@ func TestReserveScheduledExecutorTimesOutWhenNoExecutor(t *testing.T) {
 
 	start := time.Now()
 	pluginSvc.Shutdown()
-	_, _, err = pluginSvc.reserveScheduledExecutor(context.Background(), "missing-job-type", policy, nil)
+	_, _, err = pluginSvc.reserveScheduledExecutor(context.Background(), "missing-job-type", policy)
 	if err == nil {
 		t.Fatalf("expected reservation shutdown error")
 	}
@@ -303,7 +297,7 @@ func TestReserveScheduledExecutorWaitsForWorkerCapacity(t *testing.T) {
 		ExecutorReserveBackoff: 5 * time.Millisecond,
 	}
 
-	_, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
+	_, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
 	if err != nil {
 		t.Fatalf("reserve executor 1: %v", err)
 	}
@@ -314,7 +308,7 @@ func TestReserveScheduledExecutorWaitsForWorkerCapacity(t *testing.T) {
 	}
 	secondReserveCh := make(chan reserveResult, 1)
 	go func() {
-		_, release2, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil)
+		_, release2, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy)
 		if release2 != nil {
 			release2()
 		}
@@ -998,50 +992,3 @@ func TestLaneStatus_LifecycleNextDetectionSurvivesDefaultLanePrune(t *testing.T)
 			status.NextDetectionAt, expected)
 	}
 }
-
-func TestReserveWaterfillsPreferredHosts(t *testing.T) {
-	t.Parallel()
-
-	pluginSvc, err := New(Options{})
-	if err != nil {
-		t.Fatalf("New: %v", err)
-	}
-	defer pluginSvc.Shutdown()
-
-	ids := []string{"worker-a", "worker-b", "worker-c", "worker-d"}
-	for _, id := range ids {
-		pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
-			WorkerId: id,
-			Address:  "10.0.0.1:9334",
-			Capabilities: []*plugin_pb.JobTypeCapability{
-				{JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 8},
-			},
-		})
-	}
-
-	policy := schedulerPolicy{
-		PerWorkerConcurrency:   8,
-		ExecutorReserveBackoff: time.Millisecond,
-		LocalQueuePin:          5,
-	}
-	preferred := []string{"10.0.0.1:8080"}
-
-	seen := map[string]int{}
-	var releases []func()
-	defer func() {
-		for _, release := range releases {
-			release()
-		}
-	}()
-	for i := 0; i < 4; i++ {
-		exec, release, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred)
-		if err != nil {
-			t.Fatalf("reserve %d: %v", i, err)
-		}
-		releases = append(releases, release)
-		seen[exec.WorkerID]++
-	}
-	if len(seen) != 4 {
-		t.Fatalf("want one job per worker, got %v", seen)
-	}
-}

weed/admin/plugin/registry.go

@@ -277,24 +277,14 @@ func (r *Registry) PickDetector(jobType string) (*WorkerSession, error) {
 }
 
 // PickExecutor picks one executor worker for a job type.
-// 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
+func (r *Registry) PickExecutor(jobType string) (*WorkerSession, error) {
+	return r.pickByKind(jobType, false)
 }
 
 // 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.
-// 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) {
+func (r *Registry) ListExecutors(jobType string) ([]*WorkerSession, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -303,16 +293,8 @@ func (r *Registry) ListExecutors(jobType string, preferredHosts ...string) ([]*W
 		return nil, fmt.Errorf("no executor worker available for job_type=%s", jobType)
 	}
 
-	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)
-	}
+	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,70 +401,3 @@ 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/admin/plugin/worker_job_queues.go

@@ -1,317 +0,0 @@
-package plugin
-
-import (
-	"sync"
-	"time"
-
-	"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.
-// leftover is the homeless FIFO: replica hostnames with no currently
-// registered worker. Orphan grace does not apply here.
-type workerJobQueues struct {
-	mu           sync.Mutex
-	q            map[string][]*queuedJob
-	leftover     []*queuedJob
-	remaining    int
-	missingSince map[string]time.Time
-	homes        map[string]struct{}
-}
-
-func newWorkerJobQueues() *workerJobQueues {
-	return &workerJobQueues{
-		q:            make(map[string][]*queuedJob),
-		missingSince: make(map[string]time.Time),
-		homes:        make(map[string]struct{}),
-	}
-}
-
-func (qs *workerJobQueues) hasHome(workerID string) bool {
-	if qs == nil || workerID == "" {
-		return false
-	}
-	_, ok := qs.homes[workerID]
-	return ok
-}
-
-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) leftoverLen() int {
-	if qs == nil {
-		return 0
-	}
-	qs.mu.Lock()
-	defer qs.mu.Unlock()
-	return len(qs.leftover)
-}
-
-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. A home exists only if a currently registered worker's -address
-// hostname matches a replica host. Otherwise the job is homeless: global
-// leftover FIFO, no fake home, no K, no orphan grace. Single-threaded.
-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, _ := partitionByPreferredHost(executors, JobReplicaHosts(job))
-		if len(pref) == 0 {
-			// Never registered for that hostname: homeless, not an orphan.
-			qs.push("", job)
-			continue
-		}
-		for _, e := range pref {
-			if e != nil && e.WorkerID != "" {
-				qs.homes[e.WorkerID] = struct{}{}
-			}
-		}
-		sortExecutorsByLoad(pref, func(e *WorkerSession) int {
-			return r.executorReservationLoad(e, lane) + queued[e.WorkerID]
-		})
-		home := pref[0].WorkerID
-		qs.push(home, job)
-		queued[home]++
-		r.bindQueuedJobWorker(job, home)
-	}
-	return qs
-}
-
-// takeQueuedJob pops a job for a free worker.
-// Home worker (hostname matched a replica / has a home-queue identity):
-//  1. own home queue
-//  2. steal from another home if len>=K, or orphan past OrphanRehomeGrace
-//  3. homeless leftover FIFO
-// No-home worker (no matching replica, no home-queue identity):
-//  1. homeless leftover FIFO
-//  2. only if leftover empty: steal (K / orphan), same rules
-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
-	defer func() {
-		if item != nil && targetID != "" {
-			r.bindQueuedJobWorker(item.job, targetID)
-		}
-	}()
-
-	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, ""
-	}
-
-	takeLeftover := func() (*queuedJob, string) {
-		if len(qs.leftover) == 0 {
-			return nil, ""
-		}
-		item, qs.leftover = popFront(qs.leftover)
-		qs.remaining--
-		return item, thief.WorkerID
-	}
-	if !qs.hasHome(thief.WorkerID) {
-		if item, id := takeLeftover(); item != nil {
-			return item, id
-		}
-	}
-
-	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]
-		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
-		}
-		if !better && orphan == victimOrphan {
-			better = len(qq) > victimLen || (len(qq) == victimLen && id < victimID)
-		}
-		if better {
-			victimID = id
-			victimLen = len(qq)
-			victimOrphan = orphan
-		}
-	}
-	if victimID != "" {
-		item, qs.q[victimID] = popFront(qs.q[victimID])
-		qs.remaining--
-		return item, thief.WorkerID
-	}
-	return takeLeftover()
-}

weed/admin/plugin/worker_job_queues_test.go

@@ -1,600 +0,0 @@
-package plugin
-
-import (
-	"fmt"
-	"testing"
-	"time"
-
-	"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 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,
-		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"))
-	}
-
-	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("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"))
-	}
-
-	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 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()
-
-	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)
-	}
-}
-
-func TestHomelessAllRemoteWorkersUseFIFOWithoutPinOrGrace(t *testing.T) {
-	t.Parallel()
-
-	pluginSvc, err := New(Options{})
-	if err != nil {
-		t.Fatalf("New: %v", err)
-	}
-	defer pluginSvc.Shutdown()
-
-	// Workers never registered on the replica hostname — homeless, not orphan.
-	registerECWorkers(t, pluginSvc, map[string]string{
-		"worker-a": "10.0.0.5:9334",
-		"worker-b": "10.0.0.6:9334",
-	})
-
-	jobs := []*plugin_pb.JobSpec{
-		replicaJob("h1", "10.0.0.1:8080"),
-		replicaJob("h2", "10.0.0.1:8080"),
-		replicaJob("h3", "10.0.0.1:8080"),
-	}
-	policy := schedulerPolicy{
-		PerWorkerConcurrency: 1,
-		LocalQueuePin:        5,
-		OrphanRehomeGrace:    30 * time.Second,
-	}
-	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
-	if qs.queueLen("worker-a") != 0 || qs.queueLen("worker-b") != 0 {
-		t.Fatalf("homeless jobs must not get a fake home, a=%d b=%d", qs.queueLen("worker-a"), qs.queueLen("worker-b"))
-	}
-	if qs.leftoverLen() != 3 {
-		t.Fatalf("want 3 homeless FIFO, got %d", qs.leftoverLen())
-	}
-
-	item1, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item1 == nil || item1.job.JobId != "h1" {
-		t.Fatalf("FIFO first want h1, got %v", item1)
-	}
-	_, 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 || item2.job.JobId != "h2" {
-		t.Fatalf("FIFO second want h2 immediately, got %v", item2)
-	}
-	if target2 == "" || target2 == target1 {
-		t.Fatalf("second take should hit the other idle worker, t1=%s t2=%s", target1, target2)
-	}
-	if qs.leftoverLen() != 1 {
-		t.Fatalf("want 1 homeless left, got %d", qs.leftoverLen())
-	}
-}
-
-func TestMixedReplicaLocalJobsStayHomeBeforeHomeless(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("homeless-1", "10.0.0.9:8080"),
-		replicaJob("homeless-2", "10.0.0.9:8080"),
-	}
-	policy := schedulerPolicy{
-		PerWorkerConcurrency: 1,
-		LocalQueuePin:        5,
-		OrphanRehomeGrace:    30 * time.Second,
-	}
-	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
-	if qs.queueLen("worker-local") != 2 || qs.queueLen("worker-remote") != 0 || qs.leftoverLen() != 2 {
-		t.Fatalf("homes local=%d remote=%d leftover=%d", qs.queueLen("worker-local"), qs.queueLen("worker-remote"), qs.leftoverLen())
-	}
-
-	item1, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item1 == nil || item1.job.JobId != "local-1" || target1 != "worker-local" {
-		t.Fatalf("first take want local-1 on home, got job=%v target=%s", item1, target1)
-	}
-	_, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local")
-	if err != nil {
-		t.Fatalf("reserve local: %v", err)
-	}
-	defer release()
-
-	item2, target2 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item2 == nil || item2.job.JobId != "homeless-1" || target2 != "worker-remote" {
-		t.Fatalf("remote should take homeless, not steal pinned home, got job=%v target=%s", item2, target2)
-	}
-	if qs.queueLen("worker-local") != 1 {
-		t.Fatalf("remaining replica-local job must stay home, got %d", qs.queueLen("worker-local"))
-	}
-	if qs.leftoverLen() != 1 {
-		t.Fatalf("one homeless should remain, leftover=%d", qs.leftoverLen())
-	}
-}
-
-func TestIdleHomeWorkerTakesHomelessAfterEmptyHomeQueue(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",
-	})
-
-	jobs := []*plugin_pb.JobSpec{
-		replicaJob("home-1", "10.0.0.1:8080"),
-		replicaJob("homeless-1", "10.0.0.9:8080"),
-	}
-	policy := schedulerPolicy{
-		PerWorkerConcurrency: 1,
-		LocalQueuePin:        5,
-		OrphanRehomeGrace:    30 * time.Second,
-	}
-	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
-	if qs.queueLen("worker-local") != 1 || qs.leftoverLen() != 1 {
-		t.Fatalf("home=%d leftover=%d", qs.queueLen("worker-local"), qs.leftoverLen())
-	}
-
-	item1, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item1 == nil || item1.job.JobId != "home-1" || target1 != "worker-local" {
-		t.Fatalf("first take want home-1, got job=%v target=%s", item1, target1)
-	}
-	item2, target2 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item2 == nil || item2.job.JobId != "homeless-1" || target2 != "worker-local" {
-		t.Fatalf("idle home with empty queue may take homeless, got job=%v target=%s", item2, target2)
-	}
-	if qs.leftoverLen() != 0 {
-		t.Fatalf("homeless should be gone, leftover=%d", qs.leftoverLen())
-	}
-}
-
-func TestHomelessWorkerTakesFIFOBeforeStealableHome(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, 7)
-	for i := 0; i < 6; i++ {
-		jobs = append(jobs, replicaJob(fmt.Sprintf("j%d", i), "10.0.0.1:8080"))
-	}
-	jobs = append(jobs, replicaJob("homeless-1", "10.0.0.9:8080"))
-	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5}
-	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
-	if qs.queueLen("worker-local") != 6 || qs.leftoverLen() != 1 {
-		t.Fatalf("home=%d leftover=%d", qs.queueLen("worker-local"), qs.leftoverLen())
-	}
-	if qs.hasHome("worker-remote") {
-		t.Fatal("remote must have no home identity")
-	}
-
-	_, 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()
-
-	item, takeTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item == nil || takeTarget != "worker-remote" || item.job.JobId != "homeless-1" {
-		t.Fatalf("remote must take homeless FIFO before K-steal, got job=%v target=%s", item, takeTarget)
-	}
-	if qs.leftoverLen() != 0 {
-		t.Fatalf("homeless should be gone, leftover=%d", qs.leftoverLen())
-	}
-	if qs.queueLen("worker-local") != 5 {
-		t.Fatalf("stealable home must stay until FIFO empty, got %d", qs.queueLen("worker-local"))
-	}
-
-	stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if stolen == nil || stealTarget != "worker-remote" || stolen.job.JobId == "homeless-1" {
-		t.Fatalf("after FIFO empty remote may steal >=K, got job=%v target=%s", stolen, stealTarget)
-	}
-}
-
-func TestHomeWorkerStealsBeforeHomelessAfterOwnQueueEmpty(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-other": "10.0.0.2:9334",
-	})
-
-	jobs := make([]*plugin_pb.JobSpec, 0, 8)
-	for i := 0; i < 6; i++ {
-		jobs = append(jobs, replicaJob(fmt.Sprintf("j%d", i), "10.0.0.1:8080"))
-	}
-	jobs = append(jobs, replicaJob("other-1", "10.0.0.2:8080"))
-	jobs = append(jobs, replicaJob("homeless-1", "10.0.0.9:8080"))
-	policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5}
-	qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy)
-	if qs.queueLen("worker-local") != 6 || qs.queueLen("worker-other") != 1 || qs.leftoverLen() != 1 {
-		t.Fatalf("local=%d other=%d leftover=%d", qs.queueLen("worker-local"), qs.queueLen("worker-other"), qs.leftoverLen())
-	}
-	if !qs.hasHome("worker-other") {
-		t.Fatal("other must keep home identity")
-	}
-
-	item1, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item1 == nil || target1 != "worker-local" {
-		t.Fatalf("first take want local home, got job=%v target=%s", item1, target1)
-	}
-	_, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local")
-	if err != nil {
-		t.Fatalf("reserve local: %v", err)
-	}
-	defer release()
-
-	item2, target2 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if item2 == nil || item2.job.JobId != "other-1" || target2 != "worker-other" {
-		t.Fatalf("other should drain own home first, got job=%v target=%s", item2, target2)
-	}
-
-	stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy)
-	if stolen == nil || stealTarget != "worker-other" || stolen.job.JobId == "homeless-1" {
-		t.Fatalf("home worker with empty own queue still steals before homeless, got job=%v target=%s", stolen, stealTarget)
-	}
-	if qs.leftoverLen() != 1 {
-		t.Fatalf("homeless should still wait, leftover=%d", qs.leftoverLen())
-	}
-}

weed/command/filer_remote_sync_dir.go

@@ -336,16 +336,14 @@ func processUpdateEvent(
 	}
 	oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
 	dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
-	eventSource := renameSource{path: util.NewFullPath(resp.Directory, message.OldEntry.Name), loc: oldDest, stamp: message.OldEntry.GetRemoteEntry()}
 	// sourceOf is where a remote-only entry's content is: the source its
 	// pending-rename marker names, which may be an earlier rename's old key,
 	// or else the event's old key
-	sourceOf := func(entry *filer_pb.Entry, fallback renameSource) renameSource {
+	sourceOf := func(entry *filer_pb.Entry, eventSource *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {
 		if src := string(entry.Extended[filer.ExtRemoteRenameSourceKey]); strings.HasPrefix(src, strings.TrimSuffix(mountedDir, "/")+"/") {
-			// a pending entry carries the stamp of the source it reads
-			return renameSource{path: util.FullPath(src), loc: toRemoteStorageLocation(util.FullPath(mountedDir), util.FullPath(src), remoteStorageMountLocation), stamp: entry.GetRemoteEntry()}
+			return toRemoteStorageLocation(util.FullPath(mountedDir), util.FullPath(src), remoteStorageMountLocation)
 		}
-		return fallback
+		return eventSource
 	}
 	if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) {
 		glog.V(2).Infof("skipping updating: %+v", resp)
@@ -386,13 +384,13 @@ func processUpdateEvent(
 			return err
 		}
 		if isRemoteOnly(current) {
-			source := sourceOf(current, eventSource)
+			source := sourceOf(current, oldDest)
 			if err := completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, source, dest, storageClass); err != nil {
 				return err
 			}
 			// a pending entry the filer copied on this rename leaves the
 			// source its marker named
-			if pending := sourceOf(message.OldEntry, renameSource{}); pending.loc != nil && !proto.Equal(pending.loc, source.loc) {
+			if pending := sourceOf(message.OldEntry, nil); pending != nil && !proto.Equal(pending, source) {
 				return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry)
 			}
 			return nil
@@ -418,7 +416,7 @@ func processUpdateEvent(
 	if errors.Is(writeErr, errSuperseded) {
 		glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr)
 		if !proto.Equal(oldDest, dest) {
-			if err := uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, eventSource, dest, storageClass, sourceOf); err != nil {
+			if err := uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass, sourceOf); err != nil {
 				return err
 			}
 			return releaseRenamedSource(filerSource, client, mountedDir, remoteStorageMountLocation, message)
@@ -483,14 +481,7 @@ func isRemoteOnly(entry *filer_pb.Entry) bool {
 // copied over the destination, the entry is stamped with the copy, and the
 // old key goes. With neither object present the content is lost: the event
 // fails and holds the offset for recovery.
-func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir string, current *filer_pb.Entry, source renameSource, dest *remote_pb.RemoteStorageLocation, storageClass string) error {
-	oldDest := source.loc
-	// the source's recorded stamp; a retry can find the entry already
-	// stamped with the copy, whose ETag a multipart source does not share
-	sourceStamp := source.stamp
-	if sourceStamp == nil {
-		sourceStamp = proto.Clone(current.RemoteEntry).(*filer_pb.RemoteEntry)
-	}
+func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir string, current *filer_pb.Entry, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string) error {
 	if existing, err := client.StatFile(dest); err == nil {
 		if describes(current.RemoteEntry, existing) {
 			glog.V(0).Infof("%s already holds the renamed content", remote_storage.FormatLocation(dest))
@@ -500,7 +491,7 @@ func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_st
 					return err
 				}
 			}
-			return deleteRenameSource(filerClient, client, dir, current.Name, source, sourceStamp)
+			return deleteRenameSource(filerClient, client, dir, current.Name, oldDest)
 		}
 		glog.V(0).Infof("%s holds an object the entry does not describe (size %d, want %d); replacing it from %s", remote_storage.FormatLocation(dest), existing.RemoteSize, current.RemoteEntry.RemoteSize, remote_storage.FormatLocation(oldDest))
 	} else if !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
@@ -526,42 +517,21 @@ func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_st
 	if err := updateLocalEntry(filerClient, dir, current, remoteEntry); err != nil {
 		return err
 	}
-	return deleteRenameSource(filerClient, client, dir, current.Name, source, sourceStamp)
-}
-
-// renameSource is where a renamed remote-only entry's content was read
-// from: the path, the object it maps to, and the stamp describing it.
-type renameSource struct {
-	path  util.FullPath
-	loc   *remote_pb.RemoteStorageLocation
-	stamp *filer_pb.RemoteEntry
+	return deleteRenameSource(filerClient, client, dir, current.Name, oldDest)
 }
 
 // deleteRenameSource deletes a rename's old key once the renamed entry no
 // longer reads it: an entry still carrying the pending-source marker (its
-// stamp lost a race) keeps the source, and the event is retried. The key is
-// left alone when the object there is no longer the one stamp describes, as
-// when another rename copied onto it, visible yet or not.
-func deleteRenameSource(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, source renameSource, stamp *filer_pb.RemoteEntry) error {
+// stamp lost a race) keeps the source, and the event is retried.
+func deleteRenameSource(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest *remote_pb.RemoteStorageLocation) error {
 	current, err := currentEntry(filerClient, dir, name)
 	if err != nil {
 		return err
 	}
 	if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
-		return fmt.Errorf("%s still reads %s", util.NewFullPath(dir, name), remote_storage.FormatLocation(source.loc))
+		return fmt.Errorf("%s still reads %s", util.NewFullPath(dir, name), remote_storage.FormatLocation(oldDest))
 	}
-	live, err := client.StatFile(source.loc)
-	if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-	if !describes(stamp, live) || stamp.GetRemoteETag() != "" && live.RemoteETag == "" {
-		glog.V(0).Infof("keep %s: it is no longer the renamed content", remote_storage.FormatLocation(source.loc))
-		return nil
-	}
-	return deleteRemoteFile(client, source.loc)
+	return deleteRemoteFile(client, oldDest)
 }
 
 // describes reports whether the object a stat returned is the one the entry's
@@ -597,7 +567,7 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb.
 // up once. A remote-only entry is finished the way completeRemoteOnlyRename
 // finishes a rename: its stamp can already describe the destination (a sync
 // plus remote.uncache in the meantime), and only then is it complete.
-func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldSource renameSource, dest *remote_pb.RemoteStorageLocation, storageClass string, sourceOf func(*filer_pb.Entry, renameSource) renameSource) error {
+func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string, sourceOf func(*filer_pb.Entry, *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation) error {
 	current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil
@@ -609,7 +579,7 @@ func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.F
 		return nil
 	}
 	if isRemoteOnly(current) {
-		return completeRemoteOnlyRename(filerClient, client, dir, current, sourceOf(current, oldSource), dest, storageClass)
+		return completeRemoteOnlyRename(filerClient, client, dir, current, sourceOf(current, oldDest), dest, storageClass)
 	}
 	if shouldSendToRemote(current) {
 		glog.V(0).Infof("leaving %s to the rewrite that superseded the rename", remote_storage.FormatLocation(dest))

weed/command/filer_remote_sync_dir_test.go

@@ -691,10 +691,8 @@ type recordingRemote struct {
 	// objects present on the remote, by path: StatFile and ReadFile answer
 	// from it, everything else is ErrRemoteObjectNotFound.
 	objects map[string][]byte
-	// statETag is the ETag StatFile reports for a present object, unless
-	// statETags names one for its path.
-	statETag  string
-	statETags map[string]string
+	// statETag is the ETag StatFile reports for a present object.
+	statETag string
 }
 
 func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (*filer_pb.RemoteEntry, error) {
@@ -723,11 +721,7 @@ func (r *recordingRemote) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer
 	if !ok {
 		return nil, remote_storage.ErrRemoteObjectNotFound
 	}
-	etag := r.statETag
-	if e, ok := r.statETags[loc.Path]; ok {
-		etag = e
-	}
-	return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body)), RemoteETag: etag}, nil
+	return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body)), RemoteETag: r.statETag}, nil
 }
 
 func (r *recordingRemote) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) ([]byte, error) {
@@ -1393,7 +1387,7 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) {
 		current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 1024})
 		remote := &recordingRemote{}
 		filerClient := &stubFilerClient{entry: current}
-		err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", renameSource{loc: wantDelete}, wantWrite, "", eventSource)
+		err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "", eventSource)
 		if err == nil || !strings.Contains(err.Error(), "neither") {
 			t.Errorf("err = %v, want the lost-content failure", err)
 		}
@@ -1404,9 +1398,9 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) {
 
 	t.Run("uncached with the destination already synced: complete without a write", func(t *testing.T) {
 		current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 7})
-		remote := &recordingRemote{objects: map[string][]byte{"/b/dir/a.txt": []byte("payload"), "/b/dir/b.txt": []byte("payload")}}
+		remote := &recordingRemote{objects: map[string][]byte{"/b/dir/b.txt": []byte("payload")}}
 		filerClient := &stubFilerClient{entry: current}
-		if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", renameSource{loc: wantDelete}, wantWrite, "", eventSource); err != nil {
+		if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "", eventSource); err != nil {
 			t.Fatalf("err = %v, want nil: the destination object matches the entry's stamp", err)
 		}
 		if len(remote.writes) != 0 {
@@ -1439,7 +1433,7 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
 		current := pending()
 		filerClient := &stubFilerClient{entry: current}
 		remote := &recordingRemote{objects: objects(), statETag: `"copy"`}
-		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", current, renameSource{path: "/buckets/b/src/a.jpg", loc: oldDest}, dest, ""); err != nil {
+		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", current, oldDest, dest, ""); err != nil {
 			t.Fatalf("err = %v", err)
 		}
 		if filerClient.updates != 1 {
@@ -1454,7 +1448,7 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
 		// the stamp did not land: the filer still has the marker
 		filerClient := &stubFilerClient{entry: pending()}
 		remote := &recordingRemote{objects: objects(), statETag: `"copy"`}
-		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", pending(), renameSource{path: "/buckets/b/src/a.jpg", loc: oldDest}, dest, ""); err == nil {
+		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", pending(), oldDest, dest, ""); err == nil {
 			t.Fatal("err = nil, want the event retried")
 		}
 		if len(remote.deletes) != 0 {
@@ -1464,8 +1458,8 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
 }
 
 // eventSource resolves a rename's source to the event's old key.
-func eventSource(_ *filer_pb.Entry, source renameSource) renameSource {
-	return source
+func eventSource(_ *filer_pb.Entry, oldDest *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {
+	return oldDest
 }
 
 func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) {
@@ -1511,7 +1505,7 @@ func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) {
 
 	t.Run("the second rename copies from the marker's source", func(t *testing.T) {
 		current := pendingC()
-		remote := &recordingRemote{objects: objects(), statETag: `"a"`}
+		remote := &recordingRemote{objects: objects()}
 		filerClient := &stubFilerClient{entry: current}
 		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("b.jpg", "c.jpg")); err != nil {
 			t.Fatal(err)
@@ -1717,37 +1711,3 @@ func TestMetadataOnlyUpdateRestampsSyncTs(t *testing.T) {
 		t.Error("the restamped entry should read synced")
 	}
 }
-
-func TestCompleteRemoteOnlyRenameKeepsAReusedOldKey(t *testing.T) {
-	// A was renamed to B, then D to A before the sync got to the first
-	// rename: the object at A is now D's, and the only copy of it
-	oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/a.jpg"}
-	dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/b.jpg"}
-	current := &filer_pb.Entry{Name: "b.jpg", RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"a"`, RemoteSize: 6}}
-	remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-d"), "/b/b.jpg": []byte("data-a")}, statETag: `"a"`}
-	remote.statETags = map[string]string{"/b/a.jpg": `"d"`}
-	filerClient := &stubFilerClient{entry: current}
-	if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b", current, renameSource{path: "/buckets/b/a.jpg", loc: oldDest}, dest, ""); err != nil {
-		t.Fatal(err)
-	}
-	if len(remote.deletes) != 0 {
-		t.Errorf("deletes = %+v, want the reused key kept", remote.deletes)
-	}
-}
-
-func TestCompleteRemoteOnlyRenameRetryUsesTheSourceStamp(t *testing.T) {
-	// an earlier run copied a multipart source and stamped the entry with
-	// the copy's ETag, then failed to delete the source
-	oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/a.jpg"}
-	dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/b.jpg"}
-	current := &filer_pb.Entry{Name: "b.jpg", RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"copy"`, RemoteSize: 6}}
-	remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("data-a")},
-		statETags: map[string]string{"/b/a.jpg": `"a-2"`, "/b/b.jpg": `"copy"`}}
-	source := renameSource{path: "/buckets/b/a.jpg", loc: oldDest, stamp: &filer_pb.RemoteEntry{RemoteETag: `"a-2"`, RemoteSize: 6}}
-	if err := completeRemoteOnlyRename(&stubFilerClient{entry: current}, remote, "/buckets/b", current, source, dest, ""); err != nil {
-		t.Fatal(err)
-	}
-	if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/a.jpg" {
-		t.Errorf("deletes = %+v, want the unchanged source deleted on the retry", remote.deletes)
-	}
-}

weed/command/worker_runtime.go

@@ -82,7 +82,7 @@ func runPluginWorkerWithOptions(options pluginWorkerRunOptions) bool {
 		return false
 	}
 
-	handlers, err := buildPluginWorkerHandlers(options.JobTypes, dialOption, options.MaxExecute, options.WorkingDir, options.Address)
+	handlers, err := buildPluginWorkerHandlers(options.JobTypes, dialOption, options.MaxExecute, options.WorkingDir)
 	if err != nil {
 		glog.Errorf("Failed to build plugin worker handlers: %v", err)
 		return false
@@ -141,7 +141,7 @@ func resolvePluginWorkerID(explicitID string, workingDir string) (string, error)
 // buildPluginWorkerHandlers resolves the comma-separated jobTypes string
 // (which may contain category names like "all", "default", "heavy" and/or
 // explicit job type names/aliases) into a deduplicated slice of JobHandlers.
-func buildPluginWorkerHandlers(jobTypes string, dialOption grpc.DialOption, maxExecute int, workingDir string, workerAddress ...string) ([]pluginworker.JobHandler, error) {
+func buildPluginWorkerHandlers(jobTypes string, dialOption grpc.DialOption, maxExecute int, workingDir string) ([]pluginworker.JobHandler, error) {
 	jobTypes = strings.TrimSpace(jobTypes)
 	if jobTypes == "" {
 		jobTypes = defaultPluginWorkerJobTypes
@@ -152,16 +152,10 @@ func buildPluginWorkerHandlers(jobTypes string, dialOption grpc.DialOption, maxE
 		return nil, err
 	}
 
-	addr := ""
-	if len(workerAddress) > 0 {
-		addr = strings.TrimSpace(workerAddress[0])
-	}
-
 	opts := pluginworker.HandlerBuildOptions{
 		GrpcDialOption: dialOption,
 		MaxExecute:     maxExecute,
 		WorkingDir:     workingDir,
-		WorkerAddress:  addr,
 	}
 
 	handlers := make([]pluginworker.JobHandler, 0, len(factories))

weed/filer/filechunk_manifest.go

@@ -458,11 +458,11 @@ func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrin
 				localProcessed += writtenCount
 				totalWritten += writtenCount
 			})
-			if writeErr != nil {
-				err = writeErr
+			if !shouldRetry {
 				break
 			}
-			if !shouldRetry {
+			if writeErr != nil {
+				err = writeErr
 				break
 			}
 			if err != nil {

weed/filer/filechunk_manifest_test.go

@@ -787,23 +787,3 @@ func TestFetchWholeChunkUsesProvidedFilerJwt(t *testing.T) {
 		return ""
 	}))
 }
-
-// A completed fetch whose destination write failed is a failed copy, not a
-// short success: the caller must see the write error, or a temp disk filling
-// mid-copy would still upload the truncated file.
-func TestRetriedStreamFetchChunkDataReportsWriteError(t *testing.T) {
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.Write([]byte("0123456789"))
-	}))
-	t.Cleanup(srv.Close)
-
-	written, err := retriedStreamFetchChunkData(context.Background(), &failOnWrite{}, []string{srv.URL + "/5,abc"}, "", nil, false, true, 0, 0, nil)
-	assert.Error(t, err)
-	assert.Zero(t, written)
-}
-
-type failOnWrite struct{}
-
-func (w *failOnWrite) Write(p []byte) (int, error) {
-	return 0, fmt.Errorf("no space left")
-}

weed/pb/master.proto

@@ -305,11 +305,6 @@ message VolumeGrowRequest {
   string data_node = 7;
   uint32 memory_map_max_size_mb = 8;
   string disk_type = 9;
-  // If > 0, grow only when the average fill ratio of the layout's writable
-  // volumes has reached this threshold (e.g. 0.6). Lets callers pre-grow on
-  // fill level with one call, instead of listing volumes to compute the
-  // ratio themselves. Ignored when there are no writable volumes.
-  float grow_avg_fill_threshold = 10;
 }
 
 message AssignResponse {
@@ -616,9 +611,4 @@ message RaftLeadershipTransferResponse {
 }
 
 message VolumeGrowResponse {
-  // Average fill ratio of the layout's writable volumes at request time
-  // (0 when there are none). Reported whether or not growth was triggered.
-  float avg_fill = 1;
-  // Number of writable volumes at request time.
-  uint32 writable_count = 2;
 }

weed/pb/master_pb/master.pb.go

@@ -1739,13 +1739,8 @@ type VolumeGrowRequest struct {
 	DataNode            string                 `protobuf:"bytes,7,opt,name=data_node,json=dataNode,proto3" json:"data_node,omitempty"`
 	MemoryMapMaxSizeMb  uint32                 `protobuf:"varint,8,opt,name=memory_map_max_size_mb,json=memoryMapMaxSizeMb,proto3" json:"memory_map_max_size_mb,omitempty"`
 	DiskType            string                 `protobuf:"bytes,9,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
-	// If > 0, grow only when the average fill ratio of the layout's writable
-	// volumes has reached this threshold (e.g. 0.6). Lets callers pre-grow on
-	// fill level with one call, instead of listing volumes to compute the
-	// ratio themselves. Ignored when there are no writable volumes.
-	GrowAvgFillThreshold float32 `protobuf:"fixed32,10,opt,name=grow_avg_fill_threshold,json=growAvgFillThreshold,proto3" json:"grow_avg_fill_threshold,omitempty"`
-	unknownFields        protoimpl.UnknownFields
-	sizeCache            protoimpl.SizeCache
+	unknownFields       protoimpl.UnknownFields
+	sizeCache           protoimpl.SizeCache
 }
 
 func (x *VolumeGrowRequest) Reset() {
@@ -1841,13 +1836,6 @@ func (x *VolumeGrowRequest) GetDiskType() string {
 	return ""
 }
 
-func (x *VolumeGrowRequest) GetGrowAvgFillThreshold() float32 {
-	if x != nil {
-		return x.GrowAvgFillThreshold
-	}
-	return 0
-}
-
 type AssignResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Fid           string                 `protobuf:"bytes,1,opt,name=fid,proto3" json:"fid,omitempty"`
@@ -4671,12 +4659,7 @@ func (x *RaftLeadershipTransferResponse) GetNewLeader() string {
 }
 
 type VolumeGrowResponse struct {
-	state protoimpl.MessageState `protogen:"open.v1"`
-	// Average fill ratio of the layout's writable volumes at request time
-	// (0 when there are none). Reported whether or not growth was triggered.
-	AvgFill float32 `protobuf:"fixed32,1,opt,name=avg_fill,json=avgFill,proto3" json:"avg_fill,omitempty"`
-	// Number of writable volumes at request time.
-	WritableCount uint32 `protobuf:"varint,2,opt,name=writable_count,json=writableCount,proto3" json:"writable_count,omitempty"`
+	state         protoimpl.MessageState `protogen:"open.v1"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -4711,20 +4694,6 @@ func (*VolumeGrowResponse) Descriptor() ([]byte, []int) {
 	return file_master_proto_rawDescGZIP(), []int{68}
 }
 
-func (x *VolumeGrowResponse) GetAvgFill() float32 {
-	if x != nil {
-		return x.AvgFill
-	}
-	return 0
-}
-
-func (x *VolumeGrowResponse) GetWritableCount() uint32 {
-	if x != nil {
-		return x.WritableCount
-	}
-	return 0
-}
-
 type SuperBlockExtra_ErasureCoding struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Data          uint32                 `protobuf:"varint,1,opt,name=data,proto3" json:"data,omitempty"`
@@ -5260,7 +5229,7 @@ const file_master_proto_rawDesc = "" +
 	"\x15writable_volume_count\x18\t \x01(\rR\x13writableVolumeCount\x12\x1b\n" +
 	"\tdisk_type\x18\n" +
 	" \x01(\tR\bdiskType\x12,\n" +
-	"\x12expected_data_size\x18\v \x01(\x04R\x10expectedDataSize\"\xf5\x02\n" +
+	"\x12expected_data_size\x18\v \x01(\x04R\x10expectedDataSize\"\xbe\x02\n" +
 	"\x11VolumeGrowRequest\x122\n" +
 	"\x15writable_volume_count\x18\x01 \x01(\rR\x13writableVolumeCount\x12 \n" +
 	"\vreplication\x18\x02 \x01(\tR\vreplication\x12\x1e\n" +
@@ -5273,9 +5242,7 @@ const file_master_proto_rawDesc = "" +
 	"\x04rack\x18\x06 \x01(\tR\x04rack\x12\x1b\n" +
 	"\tdata_node\x18\a \x01(\tR\bdataNode\x122\n" +
 	"\x16memory_map_max_size_mb\x18\b \x01(\rR\x12memoryMapMaxSizeMb\x12\x1b\n" +
-	"\tdisk_type\x18\t \x01(\tR\bdiskType\x125\n" +
-	"\x17grow_avg_fill_threshold\x18\n" +
-	" \x01(\x02R\x14growAvgFillThreshold\"\xc4\x01\n" +
+	"\tdisk_type\x18\t \x01(\tR\bdiskType\"\xc4\x01\n" +
 	"\x0eAssignResponse\x12\x10\n" +
 	"\x03fid\x18\x01 \x01(\tR\x03fid\x12\x14\n" +
 	"\x05count\x18\x04 \x01(\x04R\x05count\x12\x14\n" +
@@ -5510,10 +5477,8 @@ const file_master_proto_rawDesc = "" +
 	"\x1eRaftLeadershipTransferResponse\x12'\n" +
 	"\x0fprevious_leader\x18\x01 \x01(\tR\x0epreviousLeader\x12\x1d\n" +
 	"\n" +
-	"new_leader\x18\x02 \x01(\tR\tnewLeader\"V\n" +
-	"\x12VolumeGrowResponse\x12\x19\n" +
-	"\bavg_fill\x18\x01 \x01(\x02R\aavgFill\x12%\n" +
-	"\x0ewritable_count\x18\x02 \x01(\rR\rwritableCount2\xf1\x12\n" +
+	"new_leader\x18\x02 \x01(\tR\tnewLeader\"\x14\n" +
+	"\x12VolumeGrowResponse2\xf1\x12\n" +
 	"\aSeaweed\x12I\n" +
 	"\rSendHeartbeat\x12\x14.master_pb.Heartbeat\x1a\x1c.master_pb.HeartbeatResponse\"\x00(\x010\x01\x12X\n" +
 	"\rKeepConnected\x12\x1f.master_pb.KeepConnectedRequest\x1a .master_pb.KeepConnectedResponse\"\x00(\x010\x01\x12Q\n" +

weed/plugin/worker/handler_registry.go

@@ -37,9 +37,6 @@ type HandlerBuildOptions struct {
 	GrpcDialOption grpc.DialOption
 	MaxExecute     int
 	WorkingDir     string
-	// WorkerAddress is the advertised weed worker -address (host:port).
-	// It is not a volume-server URL; handlers match replicas by hostname.
-	WorkerAddress string
 }
 
 var (

weed/server/filer_grpc_server_rename_remote_test.go

@@ -413,18 +413,3 @@ func TestTaggingCannotSetThePendingSourceMarker(t *testing.T) {
 	assert.Equal(t, "2026", string(x.Extended["Seaweed-Album"]))
 	assert.Empty(t, x.Extended[filer.ExtRemoteRenameSourceKey], "only the filer sets the pending source")
 }
-
-func TestDeletingTagsKeepsThePendingSourceMarker(t *testing.T) {
-	server, store := newRenameRemoteTestServer(t, &renameRemoteClient{})
-	entry := remoteOnlyRenameEntry("/buckets/b/x.jpg", 101)
-	entry.Extended = map[string][]byte{"Seaweed-Album": []byte("2026"), filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/src/x.jpg")}
-	store.entries["/buckets/b/x.jpg"] = entry
-
-	// no tag names: delete every tag
-	server.DeleteTaggingHandler(httptest.NewRecorder(), httptest.NewRequest(http.MethodDelete, "/buckets/b/x.jpg?tagging", nil))
-
-	x, err := store.FindEntry(context.Background(), "/buckets/b/x.jpg")
-	require.NoError(t, err)
-	assert.Empty(t, x.Extended["Seaweed-Album"])
-	assert.Equal(t, "/buckets/b/src/x.jpg", string(x.Extended[filer.ExtRemoteRenameSourceKey]), "the marker is not a tag")
-}

weed/server/filer_server_handlers_tagging.go

@@ -90,7 +90,7 @@ func (fs *FilerServer) DeleteTaggingHandler(w http.ResponseWriter, r *http.Reque
 	// delete all tags or specific tags
 	hasDeletion := false
 	for header, _ := range existingEntry.Extended {
-		if strings.HasPrefix(header, needle.PairNamePrefix) && header != filer.ExtRemoteRenameSourceKey {
+		if strings.HasPrefix(header, needle.PairNamePrefix) {
 			if len(deletions) == 0 {
 				delete(existingEntry.Extended, header)
 				hasDeletion = true

weed/server/master_grpc_server_volume.go

@@ -509,27 +509,6 @@ func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGro
 		Force:  true,
 		Reason: "grpc volume grow",
 	}
-
-	// Report the layout's fill observed at request time whether or not
-	// growth was triggered, so callers can tell an empty layout from a
-	// skipped one. A missing collection/layout reports zeros.
-	var avgFill float64
-	var writableCount int
-	if vl := ms.Topo.FindVolumeLayout(req.Collection, replicaPlacement, ttl, types.ToDiskType(req.DiskType)); vl != nil {
-		avgFill, writableCount = vl.WritableAvgFill(&volumeGrowOption)
-	}
-
-	// Conditional grow: with a threshold set, grow only when the layout's
-	// writable volumes have reached this average fill. A layout with no
-	// writable volumes (none yet, or all just filled) skips the check — that
-	// is the case growth exists to fix.
-	if req.GrowAvgFillThreshold > 0 && writableCount > 0 && avgFill < float64(req.GrowAvgFillThreshold) {
-		return &master_pb.VolumeGrowResponse{
-			AvgFill:       float32(avgFill),
-			WritableCount: uint32(writableCount),
-		}, nil
-	}
-
 	replicaCount := int64(req.WritableVolumeCount * uint32(replicaPlacement.GetCopyCount()))
 
 	if ms.Topo.AvailableSpaceFor(&volumeGrowOption) < replicaCount {
@@ -538,8 +517,5 @@ func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGro
 
 	ms.DoAutomaticVolumeGrow(&volumeGrowRequest)
 
-	return &master_pb.VolumeGrowResponse{
-		AvgFill:       float32(avgFill),
-		WritableCount: uint32(writableCount),
-	}, nil
+	return &master_pb.VolumeGrowResponse{}, nil
 }

weed/sftpd/sftp_copy_existing_test.go

@@ -1,210 +0,0 @@
-package sftpd
-
-import (
-	"bytes"
-	"context"
-	"net"
-	"net/http"
-	"net/http/httptest"
-	"testing"
-
-	"github.com/seaweedfs/seaweedfs/weed/pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	"github.com/seaweedfs/seaweedfs/weed/sftpd/user"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/status"
-)
-
-type copyTestFiler struct {
-	filer_pb.UnimplementedSeaweedFilerServer
-	entries   map[string]*filer_pb.Entry
-	volumeUrl string
-}
-
-func (f *copyTestFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
-	if entry, ok := f.entries[req.Directory+"/"+req.Name]; ok {
-		return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil
-	}
-	return nil, status.Error(codes.NotFound, "no entry is found")
-}
-
-func (f *copyTestFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
-	resp := &filer_pb.LookupVolumeResponse{LocationsMap: make(map[string]*filer_pb.Locations)}
-	for _, vid := range req.VolumeIds {
-		resp.LocationsMap[vid] = &filer_pb.Locations{Locations: []*filer_pb.Location{{Url: f.volumeUrl}}}
-	}
-	return resp, nil
-}
-
-func newCopyTestServer(t *testing.T, volumeUrl string, entries map[string]*filer_pb.Entry) *SftpServer {
-	lis, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatalf("listen: %v", err)
-	}
-	grpcServer := grpc.NewServer()
-	filer_pb.RegisterSeaweedFilerServer(grpcServer, &copyTestFiler{entries: entries, volumeUrl: volumeUrl})
-	go func() { _ = grpcServer.Serve(lis) }()
-	t.Cleanup(grpcServer.Stop)
-
-	_, port, _ := net.SplitHostPort(lis.Addr().String())
-	return &SftpServer{
-		filerAddr:      pb.ServerAddress("127.0.0.1:0." + port),
-		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
-		user:           &user.User{Username: "alice", Uid: 1000, HomeDir: "/"},
-	}
-}
-
-func chunkedEntry(size int64, chunks ...*filer_pb.FileChunk) *filer_pb.Entry {
-	return &filer_pb.Entry{
-		Attributes:  &filer_pb.FuseAttributes{FileSize: uint64(size)},
-		Chunks:      chunks,
-		IsDirectory: false,
-	}
-}
-
-func fileChunk(fileId string, offset, size int64) *filer_pb.FileChunk {
-	return &filer_pb.FileChunk{FileId: fileId, Offset: offset, Size: uint64(size), ModifiedTsNs: 1}
-}
-
-// A non-truncating write must keep untouched bytes, including a sparse file's
-// zeros between chunks. Stopping at the first gap would drop the tail.
-func TestCopyExistingContentFillsSparseGaps(t *testing.T) {
-	chunkA := []byte("AAAAAAAAAA")
-	chunkB := []byte("BBBBBBBBBB")
-	objects := map[string][]byte{"3,01": chunkA, "3,02": chunkB}
-	volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		data, ok := objects[r.URL.Path[1:]]
-		if !ok {
-			http.Error(w, "not found", http.StatusNotFound)
-			return
-		}
-		_, _ = w.Write(data)
-	}))
-	t.Cleanup(volume.Close)
-
-	entries := map[string]*filer_pb.Entry{
-		"/home/sparse.bin": chunkedEntry(100, fileChunk("3,01", 0, 10), fileChunk("3,02", 90, 10)),
-	}
-	fs := newCopyTestServer(t, volume.Listener.Addr().String(), entries)
-
-	var dst bytes.Buffer
-	if err := fs.copyExistingContent("/home/sparse.bin", &dst); err != nil {
-		t.Fatalf("copyExistingContent: %v", err)
-	}
-	want := append(append(append([]byte{}, chunkA...), make([]byte, 80)...), chunkB...)
-	if !bytes.Equal(dst.Bytes(), want) {
-		t.Fatalf("copied %d bytes, want %d with a zero-filled gap; tail present: %v", dst.Len(), len(want), bytes.Contains(dst.Bytes(), chunkB))
-	}
-}
-
-// A remote-only file has no local chunks: copying nothing would upload an
-// empty replacement on close. Refusing the read keeps the remote object.
-func TestCopyExistingContentRejectsUnreadableEntry(t *testing.T) {
-	remoteOnly := chunkedEntry(0)
-	remoteOnly.RemoteEntry = &filer_pb.RemoteEntry{RemoteSize: 100}
-	entries := map[string]*filer_pb.Entry{
-		"/home/remote.bin": remoteOnly,
-	}
-	fs := newCopyTestServer(t, "", entries)
-
-	var dst bytes.Buffer
-	if err := fs.copyExistingContent("/home/remote.bin", &dst); err == nil {
-		t.Fatal("expected refusal when no local bytes exist")
-	}
-}
-
-// A locally shortened file still names its older, larger remote object. Once
-// the local entry is newer the remote size is stale: copying must not pad the
-// file back out to it.
-func TestCopyExistingContentIgnoresStaleRemoteSize(t *testing.T) {
-	chunkA := []byte("AAAAAAAAAA")
-	volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/3,01" {
-			http.Error(w, "not found", http.StatusNotFound)
-			return
-		}
-		_, _ = w.Write(chunkA)
-	}))
-	t.Cleanup(volume.Close)
-
-	entries := map[string]*filer_pb.Entry{
-		"/home/short.bin": {
-			Attributes:  &filer_pb.FuseAttributes{FileSize: 10, Mtime: 200},
-			Chunks:      []*filer_pb.FileChunk{fileChunk("3,01", 0, 10)},
-			RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100, RemoteMtime: 100},
-		},
-	}
-	fs := newCopyTestServer(t, volume.Listener.Addr().String(), entries)
-
-	var dst bytes.Buffer
-	if err := fs.copyExistingContent("/home/short.bin", &dst); err != nil {
-		t.Fatalf("copyExistingContent: %v", err)
-	}
-	if !bytes.Equal(dst.Bytes(), chunkA) {
-		t.Fatalf("copied %d bytes, want the %d local bytes only", dst.Len(), len(chunkA))
-	}
-}
-
-// A file whose size is all holes still keeps its zeros.
-func TestCopyExistingContentAllHoles(t *testing.T) {
-	entries := map[string]*filer_pb.Entry{
-		"/home/holes.bin": chunkedEntry(10),
-	}
-	fs := newCopyTestServer(t, "", entries)
-
-	var dst bytes.Buffer
-	if err := fs.copyExistingContent("/home/holes.bin", &dst); err != nil {
-		t.Fatalf("all-holes file: %v", err)
-	}
-	if dst.Len() != 10 || !bytes.Equal(dst.Bytes(), make([]byte, 10)) {
-		t.Fatalf("wrote %d bytes, want 10 zeros", dst.Len())
-	}
-}
-
-func TestCopyExistingContentInlineAndMissing(t *testing.T) {
-	entries := map[string]*filer_pb.Entry{
-		"/home/inline.txt": {Attributes: &filer_pb.FuseAttributes{FileSize: 5}, Content: []byte("hello")},
-	}
-	fs := newCopyTestServer(t, "", entries)
-
-	var dst bytes.Buffer
-	if err := fs.copyExistingContent("/home/inline.txt", &dst); err != nil {
-		t.Fatalf("inline content: %v", err)
-	}
-	if got := dst.String(); got != "hello" {
-		t.Fatalf("inline content = %q, want %q", got, "hello")
-	}
-
-	dst.Reset()
-	if err := fs.copyExistingContent("/home/missing.txt", &dst); err != nil || dst.Len() != 0 {
-		t.Fatalf("missing file: err=%v written=%d", err, dst.Len())
-	}
-}
-
-func TestCopyExistingContentTrailingGap(t *testing.T) {
-	chunkA := []byte("AAAAAAAAAA")
-	volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/3,01" {
-			http.Error(w, "not found", http.StatusNotFound)
-			return
-		}
-		_, _ = w.Write(chunkA)
-	}))
-	t.Cleanup(volume.Close)
-
-	entries := map[string]*filer_pb.Entry{
-		"/home/trailing.bin": chunkedEntry(50, fileChunk("3,01", 0, 10)),
-	}
-	fs := newCopyTestServer(t, volume.Listener.Addr().String(), entries)
-
-	var dst bytes.Buffer
-	if err := fs.copyExistingContent("/home/trailing.bin", &dst); err != nil {
-		t.Fatalf("copyExistingContent: %v", err)
-	}
-	want := append(append([]byte{}, chunkA...), make([]byte, 40)...)
-	if !bytes.Equal(dst.Bytes(), want) {
-		t.Fatalf("copied %d bytes, want %d with a zero-filled tail", dst.Len(), len(want))
-	}
-}

weed/sftpd/sftp_filer.go

@@ -15,7 +15,6 @@ import (
 	"time"
 
 	"github.com/pkg/sftp"
-	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -24,7 +23,6 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/sftpd/user"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
-	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"google.golang.org/grpc"
 )
 
@@ -161,14 +159,6 @@ func (fs *SftpServer) newFileWriter(r *sftp.Request) (io.WriterAt, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create temp file: %w", err)
 	}
-	// Without O_TRUNC the client writes into the existing file, so start from its current bytes.
-	if !r.Pflags().Trunc {
-		if err := fs.copyExistingContent(absPath, tmpFile); err != nil {
-			tmpFile.Close()
-			os.Remove(tmpFile.Name())
-			return nil, err
-		}
-	}
 
 	return &SeaweedSftpFileWriter{
 		fs:          *fs,
@@ -182,75 +172,6 @@ func (fs *SftpServer) newFileWriter(r *sftp.Request) (io.WriterAt, error) {
 	}, nil
 }
 
-func (fs *SftpServer) copyExistingContent(absPath string, dst io.Writer) error {
-	entry, err := fs.getEntry(absPath)
-	if err == os.ErrNotExist {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-	if entry.IsDirectory {
-		return nil
-	}
-	size := int64(filer.FileSize(entry))
-	remote := entry.GetRemoteEntry()
-	if len(entry.Content) == 0 && len(entry.GetChunks()) == 0 {
-		if remote != nil && remote.RemoteSize > 0 {
-			return fmt.Errorf("existing content of %s is not readable", absPath)
-		}
-		if size == 0 {
-			return nil
-		}
-		return writeZeros(dst, size)
-	}
-	if size == 0 {
-		return nil
-	}
-	if len(entry.Content) > 0 {
-		if _, err := dst.Write(entry.Content); err != nil {
-			return err
-		}
-		if remaining := size - int64(len(entry.Content)); remaining > 0 {
-			return writeZeros(dst, remaining)
-		}
-		return nil
-	}
-	streamFn, err := filer.PrepareStreamContent(lookupFileIdFn(filer.LookupFn(fs)), filer.JwtForVolumeServer, entry.GetChunks(), 0, size)
-	if err != nil {
-		return fmt.Errorf("read existing content of %s: %w", absPath, err)
-	}
-	if err := streamFn(dst); err != nil {
-		return fmt.Errorf("read existing content of %s: %w", absPath, err)
-	}
-	return nil
-}
-
-// lookupFileIdFn adapts a filer LookupFn to the chunk streamer's provider
-// interface. PrepareStreamContent fills unstored gaps with zeros, so a sparse
-// file's tail survives the copy.
-type lookupFileIdFn wdclient.LookupFileIdFunctionType
-
-func (fn lookupFileIdFn) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType {
-	return wdclient.LookupFileIdFunctionType(fn)
-}
-
-func writeZeros(w io.Writer, size int64) error {
-	zeros := make([]byte, 8192)
-	for size > 0 {
-		n := int64(len(zeros))
-		if size < n {
-			n = size
-		}
-		written, err := w.Write(zeros[:n])
-		size -= int64(written)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func (fs *SftpServer) removeEntry(absPath string) error {
 	return fs.deleteEntry(absPath, false)
 }
@@ -325,24 +246,21 @@ func (fs *SftpServer) listAllPages(dirPath string) (sftp.ListerAt, error) {
 	var all []os.FileInfo
 	last := ""
 	for {
-		page, lastName, received, err := fs.fetchDirectoryPage(dirPath, last)
+		page, err := fs.fetchDirectoryPage(dirPath, last)
 		if err != nil {
 			return nil, err
 		}
 		all = append(all, page...)
-		if received < defaultListLimit {
+		if len(page) < defaultListLimit {
 			break
 		}
-		last = lastName
+		last = page[len(page)-1].Name()
 	}
 	return listerat(all), nil
 }
 
-// fetchDirectoryPage returns the entries the user may list, plus the last name and count the filer sent.
-func (fs *SftpServer) fetchDirectoryPage(dirPath, start string) ([]os.FileInfo, string, int, error) {
+func (fs *SftpServer) fetchDirectoryPage(dirPath, start string) ([]os.FileInfo, error) {
 	var list []os.FileInfo
-	lastName := ""
-	received := 0
 	err := fs.callWithClient(true, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
 		stream, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{Directory: dirPath, StartFromFileName: start, Limit: defaultListLimit})
 		if err != nil {
@@ -353,14 +271,9 @@ func (fs *SftpServer) fetchDirectoryPage(dirPath, start string) ([]os.FileInfo,
 			if err == io.EOF {
 				break
 			}
-			if err != nil {
-				return err
-			}
-			if r.Entry == nil {
+			if err != nil || r.Entry == nil {
 				continue
 			}
-			received++
-			lastName = r.Entry.Name
 			p := path.Join(dirPath, r.Entry.Name)
 			if err := fs.checkFilePermission(p, "list"); err != nil {
 				continue
@@ -369,7 +282,7 @@ func (fs *SftpServer) fetchDirectoryPage(dirPath, start string) ([]os.FileInfo,
 		}
 		return nil
 	})
-	return list, lastName, received, err
+	return list, err
 }
 
 // makeDir creates a new directory with proper permissions.

weed/sftpd/sftp_filer_test.go

@@ -1,24 +1,12 @@
 package sftpd
 
 import (
-	"context"
-	"fmt"
-	"net"
 	"net/http"
 	"net/http/httptest"
-	"os"
 	"strings"
 	"testing"
-	"time"
 
-	"github.com/pkg/sftp"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	"github.com/seaweedfs/seaweedfs/weed/sftpd/user"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/status"
 )
 
 // TestPutFileEscapesQueryInjection ensures a filename containing "?" cannot be
@@ -49,132 +37,3 @@ func TestPutFileEscapesQueryInjection(t *testing.T) {
 		t.Errorf("path not delivered literally: got %q, want %q", gotPath, malicious)
 	}
 }
-
-type fakeListFiler struct {
-	filer_pb.UnimplementedSeaweedFilerServer
-	names    []string
-	failFrom int // ListEntries fails after sending this many entries, when > 0
-}
-
-func (f *fakeListFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error {
-	sent := 0
-	for _, name := range f.names {
-		if name <= req.StartFromFileName || sent >= int(req.Limit) {
-			continue
-		}
-		if f.failFrom > 0 && sent == f.failFrom {
-			return status.Error(codes.Unavailable, "filer went away")
-		}
-		if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{}}}); err != nil {
-			return err
-		}
-		sent++
-	}
-	return nil
-}
-
-func (f *fakeListFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
-	return &filer_pb.LookupDirectoryEntryResponse{Entry: &filer_pb.Entry{Name: req.Name, Attributes: &filer_pb.FuseAttributes{}}}, nil
-}
-
-func newListTestServer(t *testing.T, filer *fakeListFiler) *SftpServer {
-	t.Helper()
-	lis, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatal(err)
-	}
-	srv := grpc.NewServer()
-	filer_pb.RegisterSeaweedFilerServer(srv, filer)
-	go srv.Serve(lis)
-	t.Cleanup(srv.Stop)
-	return &SftpServer{
-		filerAddr:      pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", lis.Addr().(*net.TCPAddr).Port)),
-		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
-		user: &user.User{
-			Username: "alice",
-			HomeDir:  "/",
-			Uid:      1001,
-			Gid:      1001,
-			// every entry under /d is listable except f00000
-			Permissions: map[string][]string{"/d": {"list"}, "/d/f00000": {}},
-		},
-	}
-}
-
-func listNames(t *testing.T, fs *SftpServer) ([]string, error) {
-	t.Helper()
-	type result struct {
-		lister sftp.ListerAt
-		err    error
-	}
-	done := make(chan result, 1)
-	go func() {
-		lister, err := fs.listAllPages("/d")
-		done <- result{lister, err}
-	}()
-	select {
-	case r := <-done:
-		if r.err != nil {
-			return nil, r.err
-		}
-		infos := make([]os.FileInfo, 10000)
-		n, _ := r.lister.ListAt(infos, 0)
-		var names []string
-		for _, fi := range infos[:n] {
-			names = append(names, fi.Name())
-		}
-		return names, nil
-	case <-time.After(10 * time.Second):
-		t.Fatal("listing did not return")
-		return nil, nil
-	}
-}
-
-func fileNames(n int) []string {
-	names := make([]string, n)
-	for i := range names {
-		names[i] = fmt.Sprintf("f%05d", i)
-	}
-	return names
-}
-
-func TestListAllPagesReturnsStreamError(t *testing.T) {
-	fs := newListTestServer(t, &fakeListFiler{names: fileNames(10), failFrom: 3})
-	_, err := listNames(t, fs)
-	if status.Code(err) != codes.Unavailable {
-		t.Fatalf("expected the stream error, got %v", err)
-	}
-}
-
-func TestListAllPagesContinuesPastHiddenEntries(t *testing.T) {
-	fs := newListTestServer(t, &fakeListFiler{names: fileNames(defaultListLimit + 500)})
-	names, err := listNames(t, fs)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(names) != defaultListLimit+499 {
-		t.Fatalf("expected %d entries, got %d", defaultListLimit+499, len(names))
-	}
-	if names[0] != "f00001" || names[len(names)-1] != fmt.Sprintf("f%05d", defaultListLimit+499) {
-		t.Fatalf("unexpected first/last entries %q, %q", names[0], names[len(names)-1])
-	}
-}
-
-func TestListAllPagesContinuesPastFullyHiddenPage(t *testing.T) {
-	const tail = 10
-	fs := newListTestServer(t, &fakeListFiler{names: fileNames(defaultListLimit + tail)})
-	// The whole first page is hidden from the user; only the tail is listable.
-	for i := 0; i < defaultListLimit; i++ {
-		fs.user.Permissions[fmt.Sprintf("/d/f%05d", i)] = []string{}
-	}
-	names, err := listNames(t, fs)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(names) != tail {
-		t.Fatalf("expected the %d entries after the hidden page, got %d", tail, len(names))
-	}
-	if names[0] != fmt.Sprintf("f%05d", defaultListLimit) {
-		t.Fatalf("unexpected first entry %q", names[0])
-	}
-}

weed/storage/erasure_coding/ec_volume.go

@@ -982,7 +982,7 @@ func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle.
 		return types.Offset{}, 0, nil, fmt.Errorf("FindNeedleFromEcx: %w", err)
 	}
 
-	intervals = ev.LocateEcShardNeedleInterval(version, offset.ToActualOffset(), size)
+	intervals = ev.LocateEcShardNeedleInterval(version, offset.ToActualOffset(), types.Size(needle.GetActualSize(size, version)))
 	return
 }
 

weed/storage/erasure_coding/ec_volume_test.go

@@ -256,30 +256,3 @@ func TestEcVolumeDestroySweepsBothDirs(t *testing.T) {
 		}
 	}
 }
-
-func TestLocateEcShardNeedleMatchesInterval(t *testing.T) {
-	ecxFile, err := os.OpenFile("./test_files/389.ecx", os.O_RDONLY, 0)
-	assert.NoError(t, err)
-	defer ecxFile.Close()
-
-	stat, err := ecxFile.Stat()
-	assert.NoError(t, err)
-	fileSize := stat.Size()
-
-	ev := &EcVolume{
-		VolumeId:    needle.VolumeId(389),
-		ecxFile:     ecxFile,
-		ecxFileSize: fileSize,
-		Shards: []*EcVolumeShard{
-			{ecdFileSize: 1118830592},
-		},
-		ECContext: NewDefaultECContext("", 389),
-	}
-
-	needleId, _ := types.ParseNeedleId("0f0edb92")
-	offset, size, intervals, err := ev.LocateEcShardNeedle(needleId, needle.GetCurrentVersion())
-	assert.NoError(t, err)
-
-	expectedIntervals := ev.LocateEcShardNeedleInterval(needle.GetCurrentVersion(), offset.ToActualOffset(), size)
-	assert.Equal(t, expectedIntervals, intervals)
-}

weed/topology/data_node.go

@@ -343,19 +343,6 @@ func (dn *DataNode) GetRack() *Rack {
 	return dn.Parent().(*NodeImpl).value.(*Rack)
 }
 
-func (dn *DataNode) matchesLocation(option *VolumeGrowOption) bool {
-	if option.DataCenter != "" && dn.GetDataCenter().Id() != NodeId(option.DataCenter) {
-		return false
-	}
-	if option.Rack != "" && dn.GetRack().Id() != NodeId(option.Rack) {
-		return false
-	}
-	if option.DataNode != "" && dn.Id() != NodeId(option.DataNode) {
-		return false
-	}
-	return true
-}
-
 func (dn *DataNode) GetTopology() *Topology {
 	p := dn.Parent()
 	for p.Parent() != nil {

weed/topology/topology.go

@@ -448,22 +448,6 @@ func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.Replic
 	}).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
 }
 
-// FindVolumeLayout is the read-only counterpart of GetVolumeLayout: it returns
-// nil when the collection or its layout does not exist, instead of creating
-// one. Lets callers check a layout's state without polluting the collection
-// map with an empty collection.
-func (t *Topology) FindVolumeLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) *VolumeLayout {
-	c, ok := t.collectionMap.Find(collectionName)
-	if !ok {
-		return nil
-	}
-	vl, ok := c.(*Collection).GetVolumeLayout(rp, ttl, diskType)
-	if !ok {
-		return nil
-	}
-	return vl
-}
-
 // DecayQuietVolumeSizes decays pending assign estimates across every layout.
 // A volume that changed reports within a pulse, so two quiet pulses mean the
 // size on record is the size there is.

weed/topology/volume_layout.go

@@ -650,7 +650,13 @@ func (vl *VolumeLayout) PickForWrite(count uint64, option *VolumeGrowOption) (vi
 		writableVolumeId := vl.writables[(start+i)%lenWriters]
 		volumeLocationList := vl.vid2location[writableVolumeId]
 		for _, dn := range volumeLocationList.list {
-			if !dn.matchesLocation(option) {
+			if option.DataCenter != "" && dn.GetDataCenter().Id() != NodeId(option.DataCenter) {
+				continue
+			}
+			if option.Rack != "" && dn.GetRack().Id() != NodeId(option.Rack) {
+				continue
+			}
+			if option.DataNode != "" && dn.Id() != NodeId(option.DataNode) {
 				continue
 			}
 			sample[found] = writableVolumeId
@@ -756,45 +762,6 @@ func (vl *VolumeLayout) ShouldGrowVolumes() bool {
 	return writable <= crowded
 }
 
-// WritableAvgFill returns the average fill ratio of the writable volumes and
-// how many there are. The fill of a single volume is its tracked effective
-// size over the layout's volume size limit; volumes without a size-tracking
-// entry yet (freshly created, no heartbeat size report so far) count as
-// empty. A non-empty location filter in option narrows the measure to
-// writables serving that target, so a targeted grow is not blocked or
-// triggered by fill elsewhere in the layout.
-func (vl *VolumeLayout) WritableAvgFill(option *VolumeGrowOption) (avgFill float64, writableCount int) {
-	vl.accessLock.RLock()
-	defer vl.accessLock.RUnlock()
-
-	var total uint64
-	for _, vid := range vl.writables {
-		if option != nil && (option.DataCenter != "" || option.Rack != "" || option.DataNode != "") {
-			matched := false
-			if loc := vl.vid2location[vid]; loc != nil {
-				for _, dn := range loc.list {
-					if dn.matchesLocation(option) {
-						matched = true
-						break
-					}
-				}
-			}
-			if !matched {
-				continue
-			}
-		}
-		writableCount++
-		if st := vl.sizeTracking[vid]; st != nil {
-			total += st.effectiveSize
-		}
-	}
-	if writableCount == 0 || vl.volumeSizeLimit == 0 {
-		return 0, writableCount
-	}
-	avgFill = float64(total) / (float64(writableCount) * float64(vl.volumeSizeLimit))
-	return avgFill, writableCount
-}
-
 func (vl *VolumeLayout) ShouldGrowVolumesByDcAndRack(writables *[]needle.VolumeId, dcId NodeId, rackId NodeId) bool {
 	// When replication spans multiple racks (DiffRackCount > 0), a writable
 	// volume's replicas only cover some racks in a DC. It is wrong to

weed/topology/volume_layout_avg_fill_test.go

@@ -1,179 +0,0 @@
-package topology
-
-import (
-	"math"
-	"testing"
-
-	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
-	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
-	"github.com/seaweedfs/seaweedfs/weed/storage/types"
-)
-
-func TestWritableAvgFill(t *testing.T) {
-	// 3 writable volumes at 20%, 50%, 80% fill (sizes 2000, 5000, 8000 of
-	// limit 10000): average fill is 50%.
-	layout := `
-{
-  "dc1":{
-    "rack1":{
-      "server1":{
-        "volumes":[
-          {"id":1, "size":2000, "replication":"000"},
-          {"id":2, "size":5000, "replication":"000"},
-          {"id":3, "size":8000, "replication":"000"}
-        ],
-        "limit":10
-      }
-    }
-  }
-}
-`
-	_, vl := setupPickTest(t, layout, 10000)
-
-	avgFill, writableCount := vl.WritableAvgFill(nil)
-	if writableCount != 3 {
-		t.Fatalf("expected 3 writable volumes, got %d", writableCount)
-	}
-	if math.Abs(avgFill-0.5) > 1e-9 {
-		t.Errorf("expected avg fill 0.5, got %v", avgFill)
-	}
-}
-
-func TestWritableAvgFillNoWritables(t *testing.T) {
-	layout := `
-{
-  "dc1":{
-    "rack1":{
-      "server1":{
-        "volumes":[
-          {"id":1, "size":2000, "replication":"000"}
-        ],
-        "limit":10
-      }
-    }
-  }
-}
-`
-	_, vl := setupPickTest(t, layout, 10000)
-
-	vl.accessLock.Lock()
-	vl.removeFromWritable(1)
-	vl.accessLock.Unlock()
-
-	avgFill, writableCount := vl.WritableAvgFill(nil)
-	if writableCount != 0 || avgFill != 0 {
-		t.Errorf("expected (0, 0) with no writables, got (%v, %d)", avgFill, writableCount)
-	}
-}
-
-// A writable volume without a size-tracking entry yet — freshly created, no
-// heartbeat size report so far — counts as empty, not as an error.
-func TestWritableAvgFillUntrackedVolumeCountsAsEmpty(t *testing.T) {
-	layout := `
-{
-  "dc1":{
-    "rack1":{
-      "server1":{
-        "volumes":[
-          {"id":1, "size":5000, "replication":"000"},
-          {"id":2, "size":5000, "replication":"000"}
-        ],
-        "limit":10
-      }
-    }
-  }
-}
-`
-	_, vl := setupPickTest(t, layout, 10000)
-
-	vl.accessLock.Lock()
-	delete(vl.sizeTracking, 2)
-	vl.accessLock.Unlock()
-
-	avgFill, writableCount := vl.WritableAvgFill(nil)
-	if writableCount != 2 {
-		t.Fatalf("expected 2 writable volumes, got %d", writableCount)
-	}
-	if math.Abs(avgFill-0.25) > 1e-9 {
-		t.Errorf("expected avg fill 0.25 (one of two volumes empty), got %v", avgFill)
-	}
-}
-
-func TestWritableAvgFillScopedToTarget(t *testing.T) {
-	layout := `
-{
-  "dc1":{
-    "rack1":{
-      "server1":{
-        "volumes":[
-          {"id":1, "size":9000, "replication":"000"},
-          {"id":2, "size":9000, "replication":"000"}
-        ],
-        "limit":10
-      }
-    }
-  },
-  "dc2":{
-    "rack2":{
-      "server2":{
-        "volumes":[
-          {"id":3, "size":1000, "replication":"000"}
-        ],
-        "limit":10
-      }
-    }
-  }
-}
-`
-	_, vl := setupPickTest(t, layout, 10000)
-
-	avgFill, writableCount := vl.WritableAvgFill(&VolumeGrowOption{DataCenter: "dc2"})
-	if writableCount != 1 || math.Abs(avgFill-0.1) > 1e-9 {
-		t.Fatalf("expected dc2 fill (0.1, 1), got (%v, %d)", avgFill, writableCount)
-	}
-
-	// A target with no writable volumes reports none, so growth is not
-	// blocked by fill elsewhere.
-	avgFill, writableCount = vl.WritableAvgFill(&VolumeGrowOption{DataCenter: "dc2", DataNode: "missing"})
-	if writableCount != 0 || avgFill != 0 {
-		t.Fatalf("expected (0, 0) for a target with no writables, got (%v, %d)", avgFill, writableCount)
-	}
-}
-
-func TestFindVolumeLayout(t *testing.T) {
-	layout := `
-{
-  "dc1":{
-    "rack1":{
-      "server1":{
-        "volumes":[
-          {"id":1, "size":2000, "collection":"colA", "replication":"000"}
-        ],
-        "limit":10
-      }
-    }
-  }
-}
-`
-	topo := setupWithLimit(t, layout, 10000)
-	rp, _ := super_block.NewReplicaPlacementFromString("000")
-
-	// Existing layout is found.
-	if vl := topo.FindVolumeLayout("colA", rp, needle.EMPTY_TTL, types.HardDriveType); vl == nil {
-		t.Error("expected layout for colA to be found")
-	}
-
-	// A missing collection resolves to nil and must not be created.
-	if vl := topo.FindVolumeLayout("noSuchCollection", rp, needle.EMPTY_TTL, types.HardDriveType); vl != nil {
-		t.Error("expected nil layout for unknown collection")
-	}
-	if _, ok := topo.collectionMap.Find("noSuchCollection"); ok {
-		t.Error("FindVolumeLayout must not create the collection")
-	}
-
-	// A missing layout within an existing collection also resolves to nil.
-	rpOther, _ := super_block.NewReplicaPlacementFromString("001")
-	if vl := topo.FindVolumeLayout("colA", rpOther, needle.EMPTY_TTL, types.HardDriveType); vl != nil {
-		t.Error("expected nil layout for unused replication")
-	}
-}

weed/util/net_host.go

@@ -1,79 +0,0 @@
-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

@@ -1,24 +0,0 @@
-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,7 +21,6 @@ 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"
@@ -37,9 +36,6 @@ type ErasureCodingTask struct {
 	workDir        string
 	progress       float64
 	grpcDialOption grpc.DialOption
-	// workerAddress is weed worker -address (host:port). Not a volume URL;
-	// late-bind matches replica hostnames against HostKey(workerAddress).
-	workerAddress string
 
 	// EC parameters
 	dataShards       int32
@@ -407,75 +403,9 @@ 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
 	}
-	// Late-bind from after consistency: worker -address hostname, not Sources[0]
-	// and not AddressIsLocal (replica URLs are hostnames like full-02:8081).
-	workerAddr := strings.TrimSpace(t.workerAddress)
-	if workerAddr == "" {
-		workerAddr, _ = os.Hostname()
-	}
-	if local := preferWorkerHostReplicaIfComplete(t.grpcDialOption, needle.VolumeId(t.volumeID), locs, t.server, workerAddr); local != "" && local != t.server {
-		glog.Infof("EC encode: preferring worker-host replica %s as source for volume %d (was %s)", local, t.volumeID, t.server)
-		t.server = local
-	}
 	return nil
 }
 
-type replicaFileCount struct {
-	url       string
-	fileCount uint64
-	err       error
-}
-
-// preferWorkerHostReplicaIfComplete rebinds from to a same-hostname replica
-// only when live FileCounts are equal. Diverged counts keep current (max).
-func preferWorkerHostReplicaIfComplete(dial grpc.DialOption, vid needle.VolumeId, locs []wdclient.Location, current, workerAddress string) string {
-	statuses := volume_replica.GetReplicaStatuses(dial, vid, locs)
-	replicas := make([]replicaFileCount, 0, len(statuses))
-	for _, s := range statuses {
-		replicas = append(replicas, replicaFileCount{url: s.Location.Url, fileCount: s.FileCount, err: s.Error})
-	}
-	return lateBindSourceToWorkerHost(current, workerAddress, replicas)
-}
-
-// lateBindSourceToWorkerHost: equal FileCount → replica on worker -address host
-// (any port). Diverge → keep current (already max FileCount). Worker address
-// is not a volume URL.
-func lateBindSourceToWorkerHost(current, workerAddress string, replicas []replicaFileCount) string {
-	workerHost := util.HostKey(workerAddress)
-	if workerHost == "" {
-		return current
-	}
-	var live []replicaFileCount
-	for _, r := range replicas {
-		if r.err != nil || r.url == "" {
-			continue
-		}
-		live = append(live, r)
-	}
-	if len(live) == 0 {
-		return current
-	}
-	maxCount := live[0].fileCount
-	allSame := true
-	for _, r := range live[1:] {
-		if r.fileCount != live[0].fileCount {
-			allSame = false
-		}
-		if r.fileCount > maxCount {
-			maxCount = r.fileCount
-		}
-	}
-	if !allSame {
-		return current
-	}
-	for _, r := range live {
-		if util.HostKey(r.url) == workerHost {
-			return r.url
-		}
-	}
-	return current
-}
-
 // 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/ec_task_source_host_test.go

@@ -1,31 +0,0 @@
-package erasure_coding
-
-import "testing"
-
-func TestLateBindSourceToWorkerHostEqualFileCountPrefersWorkerHost(t *testing.T) {
-	a := replicaFileCount{url: "full-01:8080", fileCount: 21}
-	b := replicaFileCount{url: "full-02:8081", fileCount: 21}
-	worker := "full-02:9334"
-
-	// Sources [A,B] — topology-first is A; worker on B must still copy B.
-	if got := lateBindSourceToWorkerHost(a.url, worker, []replicaFileCount{a, b}); got != b.url {
-		t.Fatalf("Sources [A,B] worker B: got %s want %s", got, b.url)
-	}
-	// Invert Sources — still B.
-	if got := lateBindSourceToWorkerHost(b.url, worker, []replicaFileCount{b, a}); got != b.url {
-		t.Fatalf("Sources [B,A] worker B: got %s want %s", got, b.url)
-	}
-}
-
-func TestLateBindSourceToWorkerHostDivergeKeepsMaxFileCount(t *testing.T) {
-	a := replicaFileCount{url: "full-01:8080", fileCount: 40}
-	b := replicaFileCount{url: "full-02:8081", fileCount: 21}
-	worker := "full-02:9334"
-
-	if got := lateBindSourceToWorkerHost(a.url, worker, []replicaFileCount{a, b}); got != a.url {
-		t.Fatalf("diverge [A,B]: got %s want max FileCount %s", got, a.url)
-	}
-	if got := lateBindSourceToWorkerHost(a.url, worker, []replicaFileCount{b, a}); got != a.url {
-		t.Fatalf("diverge [B,A]: got %s want max FileCount %s", got, a.url)
-	}
-}

weed/worker/tasks/erasure_coding/plugin_handler.go

@@ -26,9 +26,7 @@ func init() {
 		Category: pluginworker.CategoryHeavy,
 		Aliases:  []string{"erasure-coding", "erasure.coding", "ec"},
 		Build: func(opts pluginworker.HandlerBuildOptions) (pluginworker.JobHandler, error) {
-			h := NewErasureCodingHandler(opts.GrpcDialOption, opts.WorkingDir)
-			h.workerAddress = strings.TrimSpace(opts.WorkerAddress)
-			return h, nil
+			return NewErasureCodingHandler(opts.GrpcDialOption, opts.WorkingDir), nil
 		},
 	})
 }
@@ -41,7 +39,6 @@ type erasureCodingWorkerConfig struct {
 type ErasureCodingHandler struct {
 	grpcDialOption grpc.DialOption
 	workingDir     string
-	workerAddress  string
 }
 
 func NewErasureCodingHandler(grpcDialOption grpc.DialOption, workingDir string) *ErasureCodingHandler {
@@ -522,7 +519,6 @@ func (h *ErasureCodingHandler) Execute(
 		params.Collection,
 		h.grpcDialOption,
 	)
-	task.workerAddress = h.workerAddress
 	execCtx, execCancel := context.WithCancel(ctx)
 	defer execCancel()
 	task.SetProgressCallback(func(progress float64, stage string) {
@@ -674,15 +670,8 @@ func buildErasureCodingProposal(
 	}
 
 	sourceNode := ""
-	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)
-		}
+	if len(params.Sources) > 0 {
+		sourceNode = strings.TrimSpace(params.Sources[0].Node)
 	}
 
 	summary := fmt.Sprintf("Erasure code volume %d", result.VolumeID)
@@ -713,9 +702,6 @@ 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},
 			},