mount: let the release test's parked flush talk to a live filer

Marking the handle deleted raced with the flush's unlocked early read
of isDeleted. Pointing the fixture at a live fake filer lets the flush
complete on its own once the handle lock frees, no flag write needed.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

.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

@@ -195,7 +195,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 			return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry)
 		}
 		if filer_pb.IsDelete(resp) {
-			return processDeleteEvent(filerSource, client, mountedDir, remoteStorageMountLocation, resp)
+			return processDeleteEvent(client, mountedDir, remoteStorageMountLocation, resp)
 		}
 		if message.OldEntry != nil && message.NewEntry != nil {
 			return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp)
@@ -212,11 +212,8 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 // event the inline delete already handled), and GCS reports that case as
 // ErrRemoteObjectNotFound where S3 and Azure answer an idempotent success.
 // Returning the error would pin the sync offset on an event that has nothing
-// left to do. A file whose path a remote-only entry has taken since keeps its
-// object: a rename copied that entry's content to this key, and it exists
-// nowhere else.
+// left to do.
 func processDeleteEvent(
-	filerSource filer_pb.FilerClient,
 	client remote_storage.RemoteStorageClient,
 	mountedDir string,
 	remoteStorageMountLocation *remote_pb.RemoteStorageLocation,
@@ -235,36 +232,6 @@ func processDeleteEvent(
 	}
 	glog.V(2).Infof("delete: %+v", resp)
 	dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
-	if !message.OldEntry.IsDirectory {
-		// a deleted entry whose rename copy was still pending read its
-		// content from the source its marker names, which nothing else
-		// releases
-		if err := releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry); err != nil {
-			return err
-		}
-		// The object at the key may have been replaced since the delete: a
-		// rename copies another entry's content onto it, a rewrite uploads
-		// over it, and a copy can land before its entry is visible. Delete
-		// only an object the event's stamp still describes.
-		if live, statErr := client.StatFile(dest); statErr == nil {
-			if oldRemote := message.OldEntry.GetRemoteEntry(); oldRemote == nil ||
-				!describes(oldRemote, live) ||
-				oldRemote.RemoteETag != "" && live.RemoteETag == "" {
-				glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest))
-				return nil
-			}
-		} else if !errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) {
-			return statErr
-		}
-		current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name)
-		if err != nil {
-			return err
-		}
-		if isRemoteOnly(current) {
-			glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest))
-			return nil
-		}
-	}
 	if message.OldEntry.IsDirectory {
 		glog.V(0).Infof("rmdir  %s", remote_storage.FormatLocation(dest))
 		return client.RemoveDirectory(dest)
@@ -273,41 +240,6 @@ func processDeleteEvent(
 	return deleteRemoteFile(client, dest)
 }
 
-// releasePendingSource deletes the source a pending-rename marker on entry
-// names, once nothing needs it. The filer never deletes a marked source: a
-// concurrent rename can copy onto that key before its entry is visible. So
-// the object goes only while it is still the one entry's stamp describes, and
-// no remote-only entry holds its path.
-func releasePendingSource(filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) error {
-	src := util.FullPath(entry.GetExtended()[filer.ExtRemoteRenameSourceKey])
-	if src == "" || !strings.HasPrefix(string(src), strings.TrimSuffix(mountedDir, "/")+"/") {
-		return nil
-	}
-	srcDest := toRemoteStorageLocation(util.FullPath(mountedDir), src, remoteStorageMountLocation)
-	srcDir, srcName := src.DirAndName()
-	current, err := currentEntry(filerSource, srcDir, srcName)
-	if err != nil {
-		return err
-	}
-	if isRemoteOnly(current) {
-		glog.V(0).Infof("keep %s: a remote-only entry holds the pending rename's source", remote_storage.FormatLocation(srcDest))
-		return nil
-	}
-	live, err := client.StatFile(srcDest)
-	if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-	if stamp := entry.GetRemoteEntry(); !describes(stamp, live) || stamp.RemoteETag != "" && live.RemoteETag == "" {
-		glog.V(0).Infof("keep %s: the pending rename's source changed since", remote_storage.FormatLocation(srcDest))
-		return nil
-	}
-	glog.V(0).Infof("delete %s, the source of a pending rename", remote_storage.FormatLocation(srcDest))
-	return deleteRemoteFile(client, srcDest)
-}
-
 // deleteRemoteFile deletes the object and treats an already-absent object as
 // deleted.
 func deleteRemoteFile(client remote_storage.RemoteStorageClient, dest *remote_pb.RemoteStorageLocation) error {
@@ -336,17 +268,6 @@ 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 {
-		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 fallback
-	}
 	if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) {
 		glog.V(2).Infof("skipping updating: %+v", resp)
 		return nil
@@ -386,23 +307,7 @@ func processUpdateEvent(
 			return err
 		}
 		if isRemoteOnly(current) {
-			source := sourceOf(current, eventSource)
-			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) {
-				return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry)
-			}
-			return nil
-		}
-		if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
-			// the rename's copy was pending and the entry has moved on: a
-			// later rename carries the source in its own marker and finishes
-			// or releases it, so this one leaves it
-			glog.V(0).Infof("keep %s: the pending rename moved on", remote_storage.FormatLocation(oldDest))
-			return nil
+			return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, oldDest, dest, storageClass)
 		}
 		glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest))
 		if err := client.DeleteFile(oldDest); err != nil {
@@ -418,43 +323,14 @@ 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 {
-				return err
-			}
-			return releaseRenamedSource(filerSource, client, mountedDir, remoteStorageMountLocation, message)
+			return uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass)
 		}
 		return nil
 	}
 	if writeErr != nil {
 		return writeErr
 	}
-	if err := updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry); err != nil {
-		return err
-	}
-	if !proto.Equal(oldDest, dest) {
-		return releaseRenamedSource(filerSource, client, mountedDir, remoteStorageMountLocation, message)
-	}
-	return nil
-}
-
-// releaseRenamedSource releases a pending source the renamed entry's old
-// version read, once the destination holds the content on the remote under
-// its own key: only after the upload and its stamp, which clears the marker,
-// and only if the entry does not read through a marker any more. Released
-// earlier, an upload that fails followed by remote.uncache would leave the
-// entry reading a deleted source.
-func releaseRenamedSource(filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, message *filer_pb.EventNotification) error {
-	if len(message.OldEntry.GetExtended()[filer.ExtRemoteRenameSourceKey]) == 0 {
-		return nil
-	}
-	current, err := currentEntry(filerSource, message.NewParentPath, message.NewEntry.Name)
-	if err != nil {
-		return err
-	}
-	if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
-		return nil
-	}
-	return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry)
+	return updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry)
 }
 
 // currentEntry returns what the filer holds at dir/name now, nil when the
@@ -483,24 +359,11 @@ 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))
-			if len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
-				// reads still go to the source until the stamp clears the marker
-				if err := updateLocalEntry(filerClient, dir, current, existing); err != nil {
-					return err
-				}
-			}
-			return deleteRenameSource(filerClient, client, dir, current.Name, source, sourceStamp)
+			return deleteRemoteFile(client, 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 +389,7 @@ 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
-}
-
-// 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 {
-	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))
-	}
-	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
@@ -589,15 +417,15 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb.
 }
 
 // uploadCurrentEntry uploads what the filer holds at dir/name now, when the
-// event that superseded this rename will not. A rename whose snapshot is
-// superseded has already deleted the old key; the rewrite behind it in the
-// log uploads the destination itself unless shouldSendToRemote skips it on
-// the inherited RemoteEntry, whose RemoteMtime can equal the rewrite's mtime
-// within the same second. Only that case uploads here, so the content goes
-// 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 {
+// event that superseded this rename will not. A superseded rename has
+// already deleted the old key; the rewrite behind it in the log uploads the
+// destination itself whenever shouldSendToRemote sees it as unsynced, so
+// only a current entry that reads synced is uploaded here and the content
+// goes 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, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string) error {
 	current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil
@@ -609,7 +437,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, oldDest, dest, storageClass)
 	}
 	if shouldSendToRemote(current) {
 		glog.V(0).Infof("leaving %s to the rewrite that superseded the rename", remote_storage.FormatLocation(dest))
@@ -806,7 +634,6 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer
 	remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano()
 	expected := proto.Clone(entry).(*filer_pb.Entry)
 	entry.RemoteEntry = remoteEntry
-	delete(entry.Extended, filer.ExtRemoteRenameSourceKey)
 	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		_, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
 			Directory: dir,

weed/command/filer_remote_sync_dir_test.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -687,20 +686,12 @@ type recordingRemote struct {
 	// chunk ids of each written entry: which version of the file went up
 	writtenChunks [][]string
 	deleteErr     error
-	writeErr      error
 	// 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
 }
 
 func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (*filer_pb.RemoteEntry, error) {
-	if r.writeErr != nil {
-		return nil, r.writeErr
-	}
 	r.writes = append(r.writes, loc)
 	var ids []string
 	for _, c := range entry.GetChunks() {
@@ -723,11 +714,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))}, nil
 }
 
 func (r *recordingRemote) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) ([]byte, error) {
@@ -1021,7 +1008,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 
 	t.Run("absent object", func(t *testing.T) {
 		remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil {
+		if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err != nil {
 			t.Fatalf("err = %v, want nil: deleting an absent object is complete", err)
 		}
 		if len(remote.deletes) != 1 || !proto.Equal(remote.deletes[0], wantDelete) {
@@ -1033,7 +1020,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 		marked := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse)
 		marked.EventNotification.OldEntry.Extended = map[string][]byte{filer.ExtKeepRemoteObjectKey: []byte("true")}
 		remote := &recordingRemote{}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, marked); err != nil {
+		if err := processDeleteEvent(remote, mountedDir, mountLoc, marked); err != nil {
 			t.Fatalf("err = %v", err)
 		}
 		if len(remote.deletes) != 0 {
@@ -1041,108 +1028,9 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 		}
 	})
 
-	t.Run("a remote-only entry that took the path keeps the object", func(t *testing.T) {
-		// deleted, then a remote-only file renamed onto the path: the rename
-		// copied its content to this key before the sync got to the delete
-		remote := &recordingRemote{}
-		filerClient := &stubFilerClient{entry: &filer_pb.Entry{
-			Name:        "state.db.tmp",
-			Attributes:  &filer_pb.FuseAttributes{FileSize: 10},
-			RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10},
-		}}
-		if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 0 {
-			t.Errorf("deletes = %+v, want none", remote.deletes)
-		}
-	})
-
-	t.Run("a replaced object is kept", func(t *testing.T) {
-		// the delete recorded the old object's etag; a rename copied another
-		// entry's content onto the key since, and it must not go with it
-		deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse)
-		deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10}
-		remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`}
-		filerClient := &stubFilerClient{entry: &filer_pb.Entry{Name: "state.db.tmp", Content: []byte("copied")}}
-		if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, deleted); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 0 {
-			t.Errorf("deletes = %+v, want none", remote.deletes)
-		}
-	})
-
-	t.Run("a changed object with no entry at the path yet is kept", func(t *testing.T) {
-		// a directory rename copies a child onto the key before its new entry
-		// becomes visible; the copy may be the content's only remaining home
-		deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse)
-		deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10}
-		remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 0 {
-			t.Errorf("deletes = %+v, want none", remote.deletes)
-		}
-	})
-
-	t.Run("an unchanged object still goes", func(t *testing.T) {
-		deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse)
-		deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 3}
-		remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("old")}, statETag: `"orig"`}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 1 {
-			t.Errorf("deletes = %+v, want one", remote.deletes)
-		}
-	})
-
-	t.Run("a copy that landed before its entry is kept", func(t *testing.T) {
-		// the deleted entry was never replicated, so its event records no
-		// stamp to match the fresh rename copy waiting for its filer entry
-		remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 0 {
-			t.Errorf("deletes = %+v, want none", remote.deletes)
-		}
-	})
-
-	t.Run("an unverifiable object is kept", func(t *testing.T) {
-		// the delete recorded an etag the live stat does not report; the
-		// object cannot be confirmed as the one the delete meant
-		deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse)
-		deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 3}
-		remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("old")}}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 0 {
-			t.Errorf("deletes = %+v, want none", remote.deletes)
-		}
-	})
-
-	t.Run("an entry with local data at the path does not stop the delete", func(t *testing.T) {
-		remote := &recordingRemote{}
-		filerClient := &stubFilerClient{entry: &filer_pb.Entry{
-			Name:        "state.db.tmp",
-			Content:     []byte("local"),
-			RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10},
-		}}
-		if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil {
-			t.Fatalf("err = %v", err)
-		}
-		if len(remote.deletes) != 1 {
-			t.Errorf("deletes = %+v, want one", remote.deletes)
-		}
-	})
-
 	t.Run("other failure still fails", func(t *testing.T) {
 		remote := &recordingRemote{deleteErr: errors.New("AccessDenied: Access Denied")}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err == nil {
+		if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err == nil {
 			t.Fatal("err = nil, want the delete failure")
 		}
 	})
@@ -1393,7 +1281,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, "")
 		if err == nil || !strings.Contains(err.Error(), "neither") {
 			t.Errorf("err = %v, want the lost-content failure", err)
 		}
@@ -1404,9 +1292,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, ""); err != nil {
 			t.Fatalf("err = %v, want nil: the destination object matches the entry's stamp", err)
 		}
 		if len(remote.writes) != 0 {
@@ -1418,228 +1306,6 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) {
 	})
 }
 
-func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
-	oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/src/a.jpg"}
-	dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/dst/a.jpg"}
-	pending := func() *filer_pb.Entry {
-		return &filer_pb.Entry{
-			Name:        "a.jpg",
-			Attributes:  &filer_pb.FuseAttributes{FileSize: 4},
-			RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"copy"`, RemoteSize: 4},
-			Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/src/a.jpg")},
-		}
-	}
-	objects := func() map[string][]byte {
-		return map[string][]byte{"/b/src/a.jpg": []byte("data"), "/b/dst/a.jpg": []byte("data")}
-	}
-
-	t.Run("stamped, then the source goes", func(t *testing.T) {
-		// the destination already holds the copy; the entry still reads the
-		// source until its stamp clears the marker
-		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 {
-			t.Fatalf("err = %v", err)
-		}
-		if filerClient.updates != 1 {
-			t.Errorf("updates = %d, want the stamp that clears the marker", filerClient.updates)
-		}
-		if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/src/a.jpg" {
-			t.Errorf("deletes = %+v, want the source", remote.deletes)
-		}
-	})
-
-	t.Run("a marker still in place keeps the source", func(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 {
-			t.Fatal("err = nil, want the event retried")
-		}
-		if len(remote.deletes) != 0 {
-			t.Errorf("deletes = %+v, want none while the entry reads the source", remote.deletes)
-		}
-	})
-}
-
-// eventSource resolves a rename's source to the event's old key.
-func eventSource(_ *filer_pb.Entry, source renameSource) renameSource {
-	return source
-}
-
-func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) {
-	// A was renamed over B and B over C, both copies failing: C reads A's
-	// object through the marker, and B's key holds the old target's bytes
-	const mountedDir = "/buckets"
-	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
-	remoteOnly := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}
-	pendingC := func() *filer_pb.Entry {
-		return &filer_pb.Entry{
-			Name:        "c.jpg",
-			Attributes:  &filer_pb.FuseAttributes{FileSize: 6},
-			RemoteEntry: remoteOnly,
-			Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")},
-		}
-	}
-	event := func(oldName, newName string) *filer_pb.SubscribeMetadataResponse {
-		newEntry := pendingC()
-		newEntry.Name = newName
-		return &filer_pb.SubscribeMetadataResponse{
-			Directory: "/buckets/b",
-			EventNotification: &filer_pb.EventNotification{
-				OldEntry:      &filer_pb.Entry{Name: oldName, RemoteEntry: remoteOnly},
-				NewParentPath: "/buckets/b",
-				NewEntry:      newEntry,
-			},
-		}
-	}
-	objects := func() map[string][]byte {
-		return map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("old-c")}
-	}
-
-	t.Run("the first rename leaves the source to the entry that moved on", func(t *testing.T) {
-		remote := &recordingRemote{objects: objects()}
-		filerClient := &stubFilerClient{} // nothing at b.jpg any more
-		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("a.jpg", "b.jpg")); err != nil {
-			t.Fatal(err)
-		}
-		if len(remote.deletes) != 0 || len(remote.writes) != 0 {
-			t.Errorf("deletes = %+v, writes = %+v, want neither: c.jpg still reads a.jpg", remote.deletes, remote.writes)
-		}
-	})
-
-	t.Run("the second rename copies from the marker's source", func(t *testing.T) {
-		current := pendingC()
-		remote := &recordingRemote{objects: objects(), statETag: `"a"`}
-		filerClient := &stubFilerClient{entry: current}
-		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("b.jpg", "c.jpg")); err != nil {
-			t.Fatal(err)
-		}
-		if len(remote.written) != 1 || string(remote.written[0]) != "data-a" {
-			t.Errorf("written = %q, want a.jpg's content", remote.written)
-		}
-		if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/a.jpg" {
-			t.Errorf("deletes = %+v, want the source once the stamp cleared the marker", remote.deletes)
-		}
-	})
-}
-
-func TestPendingRenameSourceIsReleasedOnlyWhileUnchanged(t *testing.T) {
-	const mountedDir = "/buckets"
-	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
-	// b.jpg's copy failed after it replaced a target: it read a.jpg's object
-	pendingB := func() *filer_pb.Entry {
-		return &filer_pb.Entry{
-			Name:        "b.jpg",
-			RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6},
-			Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")},
-		}
-	}
-	deleted := func() *filer_pb.SubscribeMetadataResponse {
-		return &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{OldEntry: pendingB()}}
-	}
-	deletedPaths := func(r *recordingRemote) (paths []string) {
-		for _, d := range r.deletes {
-			paths = append(paths, d.Path)
-		}
-		return
-	}
-
-	t.Run("deleting the pending entry releases its source", func(t *testing.T) {
-		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a")}, statETag: `"a"`}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil {
-			t.Fatal(err)
-		}
-		if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") {
-			t.Errorf("deletes = %v, want the source", got)
-		}
-	})
-
-	t.Run("a source replaced since is kept", func(t *testing.T) {
-		// another rename copied onto a.jpg before its entry became visible
-		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-d")}, statETag: `"d"`}
-		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil {
-			t.Fatal(err)
-		}
-		if got := deletedPaths(remote); slices.Contains(got, "/b/a.jpg") {
-			t.Errorf("deletes = %v, want the replaced source kept", got)
-		}
-	})
-
-	t.Run("a destination cached since releases the source too", func(t *testing.T) {
-		// c.jpg was cached before the sync reached the rename: it is no
-		// longer remote-only, and a.jpg is still left over
-		cached := &filer_pb.Entry{Name: "c.jpg", Content: []byte("data-a"), Attributes: &filer_pb.FuseAttributes{FileSize: 6}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}}
-		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("data-a")}, statETag: `"a"`}
-		event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{
-			OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: cached,
-		}}
-		filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": cached}}
-		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event); err != nil {
-			t.Fatal(err)
-		}
-		if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") {
-			t.Errorf("deletes = %v, want the pending source released", got)
-		}
-	})
-
-	t.Run("a cached entry still reading the source keeps it until its upload lands", func(t *testing.T) {
-		// c.jpg was cached through its marker; until the upload is stamped
-		// and the marker cleared, a.jpg is what remote.uncache would fall
-		// back to
-		marked := func() *filer_pb.Entry {
-			return &filer_pb.Entry{Name: "c.jpg", Content: []byte("data-a"), Attributes: &filer_pb.FuseAttributes{FileSize: 6},
-				RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6},
-				Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")}}
-		}
-		for name, writeErr := range map[string]error{"upload fails": errors.New("upload failed"), "stamp does not land": nil} {
-			t.Run(name, func(t *testing.T) {
-				remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b")}, statETag: `"a"`, writeErr: writeErr}
-				event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{
-					OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: marked(),
-				}}
-				// the filer keeps the marker: the stub does not store updates
-				filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": marked()}}
-				_ = processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event)
-				if got := deletedPaths(remote); slices.Contains(got, "/b/a.jpg") {
-					t.Errorf("deletes = %v, want the source kept while the entry reads it", got)
-				}
-			})
-		}
-	})
-
-	t.Run("a later rename the filer copied releases the source", func(t *testing.T) {
-		// b.jpg was renamed to c.jpg and the filer copied a.jpg there at once
-		copied := &filer_pb.Entry{Name: "c.jpg", Attributes: &filer_pb.FuseAttributes{FileSize: 6}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}}
-		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("data-a")}, statETag: `"a"`}
-		event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{
-			OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: copied,
-		}}
-		filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": copied}}
-		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event); err != nil {
-			t.Fatal(err)
-		}
-		if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") {
-			t.Errorf("deletes = %v, want the pending source released", got)
-		}
-	})
-}
-
-// pathFilerClient serves entries by path; a path it lacks is not found.
-type pathFilerClient struct {
-	stubFilerClient
-	entries map[string]*filer_pb.Entry
-}
-
-func (c *pathFilerClient) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
-	return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entries[req.Directory+"/"+req.Name]}, nil
-}
-
-func (c *pathFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
-	return fn(c)
-}
-
 func TestShouldSendToRemote(t *testing.T) {
 	const mtime = 1786096669
 
@@ -1717,37 +1383,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/filer/filer_lazy_remote.go

@@ -225,97 +225,3 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool,
 	glog.V(3).InfofCtx(ctx, "maybeDeleteFromRemote: deleted %s from remote", entry.FullPath)
 	return true, nil
 }
-
-// DeleteFromRemote deletes the object, or for a directory every object, the
-// entry maps to on its mounted remote storage.
-func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error {
-	_, err := f.maybeDeleteFromRemote(ctx, entry)
-	return err
-}
-
-// ExtRemoteRenameSourceKey marks a remote-only entry whose rename committed
-// while its copy to the new key was still pending: it holds the path whose
-// object remote-only reads must serve until the copy stamps the entry.
-const ExtRemoteRenameSourceKey = "Seaweed-X-Remote-Rename-Source"
-
-// RemoteSourcePath returns the path whose remote object holds the content of
-// the remote-only entry at p: the renamed-from path a pending deferred copy
-// still reads, or p itself. A marker leading out of p's mount is ignored, so
-// whoever can set extended attributes cannot point reads at another mount.
-func (f *Filer) RemoteSourcePath(entry *Entry, p util.FullPath) util.FullPath {
-	src := util.FullPath(entry.Extended[ExtRemoteRenameSourceKey])
-	if src == "" || f.RemoteStorage == nil {
-		return p
-	}
-	mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(p)
-	if remoteLoc == nil {
-		return p
-	}
-	if srcMountDir, _ := f.RemoteStorage.FindMountDirectory(src); srcMountDir != mountDir {
-		return p
-	}
-	return src
-}
-
-// RemoteReadPath returns the dir and name whose object a remote-only read of
-// the entry at dir/name serves.
-func (f *Filer) RemoteReadPath(entry *Entry, dir, name string) (string, string) {
-	return f.RemoteSourcePath(entry, util.NewFullPath(dir, name)).DirAndName()
-}
-
-// CopyRemoteOnlyEntry copies the remote object of a renamed entry whose
-// content exists only on the remote to the key of newPath, and returns the
-// RemoteEntry of the copy. The rename deletes the old key along with the old
-// entry, and with no local data nothing would write the new key. It returns
-// nil when no copy is needed: the entry has local data, which the remote sync
-// uploads to the new key, it is empty, which the sync writes as an empty
-// object, or it is not under a remote mount. A rename that
-// moves such an entry out of its mount, or onto a remote that cannot copy, is
-// refused, since it would delete the only copy of the content.
-func (f *Filer) CopyRemoteOnlyEntry(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (*filer_pb.RemoteEntry, error) {
-	src, dst, copier, err := f.RemoteCopyLocations(ctx, entry, oldPath, newPath)
-	if err != nil || copier == nil {
-		return nil, err
-	}
-	remoteEntry, err := copier.CopyFile(src, dst)
-	if err != nil {
-		return nil, fmt.Errorf("copy remote %s to %s: %w", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst), err)
-	}
-	glog.V(3).InfofCtx(ctx, "CopyRemoteOnlyEntry: copied %s to %s", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst))
-	return remoteEntry, nil
-}
-
-// RemoteCopyLocations resolves the object locations a remote-only rename copy
-// of entry between oldPath and newPath would use, and the client able to copy
-// it. A nil copier means the entry needs no copy; an error means it needs one
-// that cannot be served, which a rename must refuse before committing.
-func (f *Filer) RemoteCopyLocations(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (src, dst *remote_pb.RemoteStorageLocation, copier remote_storage.RemoteStorageObjectCopier, err error) {
-	if f.RemoteStorage == nil || entry.IsDirectory() || !entry.IsInRemoteOnly() || len(entry.Content) > 0 {
-		return nil, nil, nil, nil
-	}
-
-	mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(oldPath)
-	if remoteLoc == nil {
-		return nil, nil, nil, nil
-	}
-	if newMountDir, _ := f.RemoteStorage.FindMountDirectory(newPath); newMountDir != mountDir {
-		return nil, nil, nil, fmt.Errorf("%s exists only on the remote of mount %s and cannot be renamed out of it; cache it first", oldPath, mountDir)
-	}
-
-	remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name)
-	if !found {
-		return nil, nil, nil, fmt.Errorf("resolve remote storage client for %s: not found", oldPath)
-	}
-	client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf)
-	if clientErr != nil {
-		return nil, nil, nil, fmt.Errorf("resolve remote storage client for %s: %w", oldPath, clientErr)
-	}
-	copier, ok := client.(remote_storage.RemoteStorageObjectCopier)
-	if !ok {
-		return nil, nil, nil, fmt.Errorf("%s exists only on the remote and remote storage type %s cannot copy it to the new name; cache it first", oldPath, remoteConf.Type)
-	}
-
-	src = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, f.RemoteSourcePath(entry, oldPath))
-	dst = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath)
-	return src, dst, copier, nil
-}

weed/mount/weedfs_file_io.go

@@ -140,6 +140,12 @@ func (wfs *WFS) invalidateOpenMtimeCache(inode uint64) {
 }
 
 func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) {
+	// Drop the file description's flock locks first: close() frees them
+	// regardless of how writeback fares, and a flush stuck on an unreachable
+	// filer must not pin a dead holder's lock in the table.
+	if in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK != 0 {
+		wfs.releaseFlockOwner(in.NodeId, in.LockOwner)
+	}
 	// Flush is usually sent before Release, but the FUSE protocol does not
 	// guarantee it. Route every Release through doFlush so a dirty handle
 	// (e.g. a deferred create with no intervening Flush) is not dropped.
@@ -153,8 +159,5 @@ func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) {
 			glog.Warningf("release fh %d inode %d: fallback flush failed: %v", in.Fh, in.NodeId, status)
 		}
 	}
-	if in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK != 0 {
-		wfs.releaseFlockOwner(in.NodeId, in.LockOwner)
-	}
 	wfs.ReleaseHandle(FileHandleId(in.Fh))
 }

weed/mount/weedfs_release_flock_test.go

@@ -0,0 +1,77 @@
+package mount
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/go-fuse/v2/fuse"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// A close() frees the file description's flock locks even when the handle
+// flush is stuck — e.g. a filer that stopped answering. If the locks outlive
+// the blocked flush, a restarted process taking the same file's flock sees
+// "file already locked" until the mount restarts.
+func TestReleaseDropsFlockWhileFlushBlocked(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	wfs.posixLocks = NewPosixLockTable()
+	startFakeFiler(t, wfs, &fakeFilerServer{})
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/f"), time.Now().Unix(), false, false, 0, false)
+	fh, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, &filer_pb.Entry{
+		Name:       "f",
+		Attributes: &filer_pb.FuseAttributes{},
+	}, 0, 0)
+	fh.dirtyMetadata = true
+
+	const owner = uint64(42)
+	if st := wfs.posixLocks.SetLk(inode, lockRange{Start: 0, End: math.MaxUint64, Typ: f_WRLCK, Owner: owner, IsFlock: true}); st != fuse.OK {
+		t.Fatalf("take flock: %v", st)
+	}
+
+	// Stall the release flush at the handle lock flushMetadataToFiler waits on.
+	active := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	released := false
+	releaseLock := func() {
+		if !released {
+			wfs.fhLockTable.ReleaseLock(fh.fh, active)
+			released = true
+		}
+	}
+	defer releaseLock()
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		wfs.Release(nil, &fuse.ReleaseIn{
+			InHeader:     fuse.InHeader{NodeId: inode},
+			Fh:           uint64(fh.fh),
+			LockOwner:    owner,
+			ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK,
+		})
+	}()
+
+	// The dead holder's flock must be gone while the flush is still parked.
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		var out fuse.LkOut
+		wfs.posixLocks.GetLk(inode, lockRange{Start: 0, End: math.MaxUint64, Typ: f_WRLCK, Owner: 7, IsFlock: true}, &out)
+		if out.Lk.Typ == f_UNLCK {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("flock still held while the release flush is blocked")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	// Let the parked flush through: the live filer answers the metadata update.
+	releaseLock()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("release never completed after the flush unblocked")
+	}
+}

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/remote_storage/remote_storage.go

@@ -110,14 +110,6 @@ type RemoteStorageStreamReader interface {
 	ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error)
 }
 
-// RemoteStorageObjectCopier is an optional interface for remote storage clients
-// that copy an object to another key within the storage, without the content
-// passing through the client. A rename of an entry whose content exists only
-// on the remote needs it.
-type RemoteStorageObjectCopier interface {
-	CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error)
-}
-
 // CacheWaitTimeout is how long a read of an uncached remote-only object waits
 // for the local cache before serving another way: small files wait longer since
 // their cache completes quickly, large files fail fast for better TTFB. A mount

weed/remote_storage/s3/s3_storage_client.go

@@ -610,7 +610,7 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL
 			copyInput := &s3.CopyObjectInput{
 				Bucket:                  aws.String(loc.Bucket),
 				Key:                     aws.String(key),
-				CopySource:              aws.String(s3CopySource(loc.Bucket, key)),
+				CopySource:              aws.String(url.PathEscape(loc.Bucket + "/" + key)),
 				MetadataDirective:       aws.String(s3.MetadataDirectiveReplace),
 				Metadata:                headOut.Metadata,
 				ContentType:             headOut.ContentType,
@@ -663,168 +663,6 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL
 	}
 	return
 }
-
-// s3CopyPartSize is the smallest part a multipart copy uses; parts grow so
-// the largest object stays within the 10000-part limit
-const s3CopyPartSize = 512 * 1024 * 1024
-
-// CopyFile copies the object at src to dst on the server side, with a single
-// CopyObject up to the 5 GB limit and a multipart copy above it. The copy
-// keeps the source's metadata.
-func (s *s3RemoteStorageClient) CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) {
-	srcKey, dstKey := src.Path[1:], dst.Path[1:]
-	headOut, err := s.conn.HeadObject(&s3.HeadObjectInput{
-		Bucket: aws.String(src.Bucket),
-		Key:    aws.String(srcKey),
-	})
-	if err != nil {
-		if reqErr, ok := err.(awserr.RequestFailure); ok && reqErr.StatusCode() == http.StatusNotFound {
-			return nil, remote_storage.ErrRemoteObjectNotFound
-		}
-		return nil, fmt.Errorf("stat s3 %s/%s before copy: %w", src.Bucket, srcKey, err)
-	}
-	copySource := aws.String(s3CopySource(src.Bucket, srcKey))
-	var storageClass *string
-	if s.conf.S3StorageClass != "" {
-		storageClass = aws.String(s.conf.S3StorageClass)
-	}
-
-	if size := aws.Int64Value(headOut.ContentLength); size <= s3CopyObjectSizeLimit {
-		// a copy takes the bucket's default encryption unless told otherwise
-		if _, err = s.conn.CopyObject(&s3.CopyObjectInput{
-			Bucket:               aws.String(dst.Bucket),
-			Key:                  aws.String(dstKey),
-			CopySource:           copySource,
-			StorageClass:         storageClass,
-			ServerSideEncryption: headOut.ServerSideEncryption,
-			SSEKMSKeyId:          headOut.SSEKMSKeyId,
-		}); err != nil {
-			return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err)
-		}
-	} else if err = s.multipartCopy(headOut, src, copySource, size, dst.Bucket, dstKey, storageClass); err != nil {
-		return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err)
-	}
-
-	return s.readFileRemoteEntry(dst)
-}
-
-// s3CopySource escapes each segment of the copy source and keeps the slashes,
-// which every S3 implementation reads back
-func s3CopySource(bucket, key string) string {
-	segments := strings.Split(bucket+"/"+key, "/")
-	for i, segment := range segments {
-		segments[i] = url.PathEscape(segment)
-	}
-	return strings.Join(segments, "/")
-}
-
-// s3TaggingHeader encodes tags as the query string x-amz-tagging takes, with
-// spaces as %20: a + is read back as a space by some implementations and
-// literally by others
-func s3TaggingHeader(tagSet []*s3.Tag) string {
-	escape := func(s string) string {
-		return strings.ReplaceAll(url.QueryEscape(s), "+", "%20")
-	}
-	pairs := make([]string, 0, len(tagSet))
-	for _, tag := range tagSet {
-		pairs = append(pairs, escape(aws.StringValue(tag.Key))+"="+escape(aws.StringValue(tag.Value)))
-	}
-	return strings.Join(pairs, "&")
-}
-
-// multipartCopy copies an object above the CopyObject limit part by part. A
-// multipart upload does not inherit the source's metadata or tags, so they are
-// carried over like CopyObject would: metadata from the source's HeadObject,
-// tags when the remote supports tagging.
-func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src *remote_pb.RemoteStorageLocation, copySource *string, size int64, bucket, key string, storageClass *string) error {
-	createInput := &s3.CreateMultipartUploadInput{
-		Bucket:                  aws.String(bucket),
-		Key:                     aws.String(key),
-		Metadata:                headOut.Metadata,
-		ContentType:             headOut.ContentType,
-		ContentEncoding:         headOut.ContentEncoding,
-		CacheControl:            headOut.CacheControl,
-		ContentDisposition:      headOut.ContentDisposition,
-		ContentLanguage:         headOut.ContentLanguage,
-		WebsiteRedirectLocation: headOut.WebsiteRedirectLocation,
-		ServerSideEncryption:    headOut.ServerSideEncryption,
-		SSEKMSKeyId:             headOut.SSEKMSKeyId,
-		StorageClass:            storageClass,
-	}
-	if headOut.Expires != nil {
-		if expires, parseErr := http.ParseTime(*headOut.Expires); parseErr == nil {
-			createInput.Expires = aws.Time(expires)
-		}
-	}
-	if s.conf.S3SupportTagging {
-		tagOut, tagErr := s.conn.GetObjectTagging(&s3.GetObjectTaggingInput{
-			Bucket: aws.String(src.Bucket),
-			Key:    aws.String(src.Path[1:]),
-		})
-		if tagErr != nil {
-			return fmt.Errorf("read tags: %w", tagErr)
-		}
-		if len(tagOut.TagSet) > 0 {
-			createInput.Tagging = aws.String(s3TaggingHeader(tagOut.TagSet))
-		}
-	}
-	createOut, err := s.conn.CreateMultipartUpload(createInput)
-	if err != nil {
-		return err
-	}
-
-	partSize := int64(s3CopyPartSize)
-	for partSize*10000 < size {
-		partSize *= 2
-	}
-	var parts []*s3.CompletedPart
-	for offset, partNumber := int64(0), int64(1); offset < size; offset, partNumber = offset+partSize, partNumber+1 {
-		partOut, partErr := s.conn.UploadPartCopy(&s3.UploadPartCopyInput{
-			Bucket:          aws.String(bucket),
-			Key:             aws.String(key),
-			UploadId:        createOut.UploadId,
-			PartNumber:      aws.Int64(partNumber),
-			CopySource:      copySource,
-			CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, min(offset+partSize, size)-1)),
-			// every part from the object the HeadObject saw, not a mix of
-			// versions if it is overwritten mid-copy
-			CopySourceIfMatch: headOut.ETag,
-		})
-		if partErr == nil && partOut.CopyPartResult == nil {
-			partErr = errors.New("no part result")
-		}
-		if partErr != nil {
-			s.abortMultipartUpload(bucket, key, createOut.UploadId)
-			return fmt.Errorf("copy part %d: %w", partNumber, partErr)
-		}
-		parts = append(parts, &s3.CompletedPart{
-			ETag:       partOut.CopyPartResult.ETag,
-			PartNumber: aws.Int64(partNumber),
-		})
-	}
-
-	if _, err = s.conn.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{
-		Bucket:          aws.String(bucket),
-		Key:             aws.String(key),
-		UploadId:        createOut.UploadId,
-		MultipartUpload: &s3.CompletedMultipartUpload{Parts: parts},
-	}); err != nil {
-		s.abortMultipartUpload(bucket, key, createOut.UploadId)
-		return err
-	}
-	return nil
-}
-
-func (s *s3RemoteStorageClient) abortMultipartUpload(bucket, key string, uploadId *string) {
-	if _, err := s.conn.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
-		Bucket:   aws.String(bucket),
-		Key:      aws.String(key),
-		UploadId: uploadId,
-	}); err != nil {
-		glog.Warningf("abort multipart copy to s3 %s/%s: %v", bucket, key, err)
-	}
-}
-
 func (s *s3RemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) {
 	_, err = s.conn.DeleteObject(&s3.DeleteObjectInput{
 		Bucket: aws.String(loc.Bucket),

weed/remote_storage/s3/s3_storage_client_copy_test.go

@@ -1,169 +0,0 @@
-package s3
-
-import (
-	"errors"
-	"net/http"
-	"testing"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/awserr"
-	awss3 "github.com/aws/aws-sdk-go/service/s3"
-	"github.com/aws/aws-sdk-go/service/s3/s3iface"
-	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
-	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
-	"github.com/stretchr/testify/require"
-)
-
-// copyObjectMock serves HeadObject for a source of size bytes and records the
-// copy calls.
-type copyObjectMock struct {
-	s3iface.S3API
-	size        int64
-	missing     bool
-	partErr     error
-	copyInputs  []*awss3.CopyObjectInput
-	partInputs  []*awss3.UploadPartCopyInput
-	createInput *awss3.CreateMultipartUploadInput
-	tagInputs   []*awss3.GetObjectTaggingInput
-	completed   *awss3.CompleteMultipartUploadInput
-	aborted     bool
-}
-
-func (m *copyObjectMock) HeadObject(input *awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) {
-	if m.missing {
-		return nil, awserr.NewRequestFailure(awserr.New("NotFound", "not found", nil), http.StatusNotFound, "")
-	}
-	return &awss3.HeadObjectOutput{
-		ContentLength:           aws.Int64(m.size),
-		ContentType:             aws.String("image/jpeg"),
-		Expires:                 aws.String("Wed, 21 Oct 2026 07:28:00 GMT"),
-		WebsiteRedirectLocation: aws.String("/elsewhere"),
-		ServerSideEncryption:    aws.String("aws:kms"),
-		SSEKMSKeyId:             aws.String("key-1"),
-		ETag:                    aws.String(`"etag-` + aws.StringValue(input.Key) + `"`),
-	}, nil
-}
-
-func (m *copyObjectMock) GetObjectTagging(input *awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
-	m.tagInputs = append(m.tagInputs, input)
-	return &awss3.GetObjectTaggingOutput{TagSet: []*awss3.Tag{
-		{Key: aws.String("album"), Value: aws.String("2026 & co")},
-		{Key: aws.String("a+b"), Value: aws.String("x=y")},
-	}}, nil
-}
-
-func (m *copyObjectMock) CopyObject(input *awss3.CopyObjectInput) (*awss3.CopyObjectOutput, error) {
-	m.copyInputs = append(m.copyInputs, input)
-	return &awss3.CopyObjectOutput{}, nil
-}
-
-func (m *copyObjectMock) CreateMultipartUpload(input *awss3.CreateMultipartUploadInput) (*awss3.CreateMultipartUploadOutput, error) {
-	m.createInput = input
-	return &awss3.CreateMultipartUploadOutput{UploadId: aws.String("upload-1")}, nil
-}
-
-func (m *copyObjectMock) UploadPartCopy(input *awss3.UploadPartCopyInput) (*awss3.UploadPartCopyOutput, error) {
-	m.partInputs = append(m.partInputs, input)
-	if m.partErr != nil && len(m.partInputs) == 2 {
-		return nil, m.partErr
-	}
-	return &awss3.UploadPartCopyOutput{CopyPartResult: &awss3.CopyPartResult{ETag: aws.String(`"part"`)}}, nil
-}
-
-func (m *copyObjectMock) CompleteMultipartUpload(input *awss3.CompleteMultipartUploadInput) (*awss3.CompleteMultipartUploadOutput, error) {
-	m.completed = input
-	return &awss3.CompleteMultipartUploadOutput{}, nil
-}
-
-func (m *copyObjectMock) AbortMultipartUpload(*awss3.AbortMultipartUploadInput) (*awss3.AbortMultipartUploadOutput, error) {
-	m.aborted = true
-	return &awss3.AbortMultipartUploadOutput{}, nil
-}
-
-func newCopyTestClient(mock *copyObjectMock) *s3RemoteStorageClient {
-	return &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock}
-}
-
-var (
-	copySrc = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/src/a b.jpg"}
-	copyDst = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dst/a b.jpg"}
-)
-
-func TestS3ClientImplementsObjectCopier(t *testing.T) {
-	var _ remote_storage.RemoteStorageObjectCopier = &s3RemoteStorageClient{}
-}
-
-func TestS3CopyFileUsesCopyObjectUpToTheLimit(t *testing.T) {
-	mock := &copyObjectMock{size: s3CopyObjectSizeLimit}
-
-	remoteEntry, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst)
-	require.NoError(t, err)
-
-	require.Len(t, mock.copyInputs, 1)
-	require.Equal(t, "bucket/src/a%20b.jpg", aws.StringValue(mock.copyInputs[0].CopySource))
-	require.Equal(t, "dst/a b.jpg", aws.StringValue(mock.copyInputs[0].Key))
-	require.Equal(t, "aws:kms", aws.StringValue(mock.copyInputs[0].ServerSideEncryption))
-	require.Equal(t, "key-1", aws.StringValue(mock.copyInputs[0].SSEKMSKeyId))
-	require.Nil(t, mock.createInput)
-	// the RemoteEntry describes the destination
-	require.Equal(t, `"etag-dst/a b.jpg"`, remoteEntry.RemoteETag)
-	require.Equal(t, int64(s3CopyObjectSizeLimit), remoteEntry.RemoteSize)
-}
-
-func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) {
-	size := int64(s3CopyObjectSizeLimit) + 1
-	mock := &copyObjectMock{size: size}
-
-	_, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst)
-	require.NoError(t, err)
-
-	require.Empty(t, mock.copyInputs)
-	require.NotNil(t, mock.createInput)
-	require.Equal(t, "image/jpeg", aws.StringValue(mock.createInput.ContentType))
-	require.Equal(t, "/elsewhere", aws.StringValue(mock.createInput.WebsiteRedirectLocation))
-	require.Equal(t, "aws:kms", aws.StringValue(mock.createInput.ServerSideEncryption))
-	require.Equal(t, "key-1", aws.StringValue(mock.createInput.SSEKMSKeyId))
-	require.Equal(t, int64(1792567680), aws.TimeValue(mock.createInput.Expires).Unix())
-	// tags only where the remote supports tagging
-	require.Empty(t, mock.tagInputs)
-	require.Nil(t, mock.createInput.Tagging)
-	require.Len(t, mock.partInputs, 11)
-	require.Equal(t, "bytes=0-536870911", aws.StringValue(mock.partInputs[0].CopySourceRange))
-	require.Equal(t, "bytes=5368709120-5368709120", aws.StringValue(mock.partInputs[10].CopySourceRange))
-	for _, part := range mock.partInputs {
-		require.Equal(t, `"etag-src/a b.jpg"`, aws.StringValue(part.CopySourceIfMatch))
-	}
-	require.Len(t, mock.completed.MultipartUpload.Parts, 11)
-	require.Equal(t, int64(11), aws.Int64Value(mock.completed.MultipartUpload.Parts[10].PartNumber))
-	require.False(t, mock.aborted)
-}
-
-func TestS3CopyFileCarriesTagsIntoAMultipartCopy(t *testing.T) {
-	mock := &copyObjectMock{size: s3CopyObjectSizeLimit + 1}
-	client := newCopyTestClient(mock)
-	client.conf.S3SupportTagging = true
-
-	_, err := client.CopyFile(copySrc, copyDst)
-	require.NoError(t, err)
-
-	require.Len(t, mock.tagInputs, 1)
-	require.Equal(t, "src/a b.jpg", aws.StringValue(mock.tagInputs[0].Key))
-	require.Equal(t, "album=2026%20%26%20co&a%2Bb=x%3Dy", aws.StringValue(mock.createInput.Tagging))
-}
-
-func TestS3CopyFileAbortsAFailedMultipartCopy(t *testing.T) {
-	mock := &copyObjectMock{size: 2 * s3CopyObjectSizeLimit, partErr: errors.New("part failed")}
-
-	_, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst)
-	require.ErrorContains(t, err, "part failed")
-	require.True(t, mock.aborted)
-	require.Nil(t, mock.completed)
-}
-
-func TestS3CopyFileMissingSource(t *testing.T) {
-	mock := &copyObjectMock{missing: true}
-
-	_, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst)
-	require.ErrorIs(t, err, remote_storage.ErrRemoteObjectNotFound)
-	require.Empty(t, mock.copyInputs)
-}

weed/server/filer_grpc_server_remote.go

@@ -96,8 +96,7 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re
 
 	glog.V(1).Infof("CacheRemoteObjectToLocalCluster: caching %s/%s (remote size: %d)", req.Directory, req.Name, entry.Remote.RemoteSize)
 
-	readDir, readName := fs.filer.RemoteReadPath(entry, req.Directory, req.Name)
-	storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, readDir, readName)
+	storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, req.Directory, req.Name)
 	if err != nil {
 		return nil, err
 	}

weed/server/filer_grpc_server_rename.go

@@ -2,9 +2,7 @@ package weed_server
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"maps"
 	"path/filepath"
 	"time"
 
@@ -47,9 +45,6 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 
 	defer fs.acquireRenamePathLocks("AtomicRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))()
 
-	// a SQL store keeps the transaction in the context and refuses calls
-	// through it once committed; the work after the commit uses this one
-	afterCommitCtx := ctx
 	ctx, err := fs.filer.BeginTransaction(ctx)
 	if err != nil {
 		return nil, err
@@ -63,8 +58,7 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 
 	var metadataEvents []metadataEvent
 	var pendingChunkDeletes []*filer_pb.FileChunk
-	var pendingRemoteCopies []*pendingRemoteCopy
-	moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies)
+	moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes)
 	if moveErr != nil {
 		fs.filer.RollbackTransaction(ctx)
 		return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
@@ -81,10 +75,8 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 	if len(pendingChunkDeletes) > 0 {
 		fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes)
 	}
-	fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies)
-	applyRemoteCopies(metadataEvents, pendingRemoteCopies)
 	for _, event := range metadataEvents {
-		event.notify(fs.filer, afterCommitCtx, req.Signatures)
+		event.notify(fs.filer, ctx, req.Signatures)
 	}
 
 	return &filer_pb.AtomicRenameEntryResponse{}, nil
@@ -105,7 +97,6 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
 
 	ctx := context.Background()
 
-	afterCommitCtx := ctx
 	ctx, err = fs.filer.BeginTransaction(ctx)
 	if err != nil {
 		return err
@@ -134,8 +125,7 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
 
 	var metadataEvents []metadataEvent
 	var pendingChunkDeletes []*filer_pb.FileChunk
-	var pendingRemoteCopies []*pendingRemoteCopy
-	moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies)
+	moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes)
 	if moveErr != nil {
 		fs.filer.RollbackTransaction(ctx)
 		return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
@@ -148,10 +138,8 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
 	if len(pendingChunkDeletes) > 0 {
 		fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes)
 	}
-	fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies)
-	applyRemoteCopies(metadataEvents, pendingRemoteCopies)
 	for _, event := range metadataEvents {
-		event.notify(fs.filer, afterCommitCtx, req.Signatures)
+		event.notify(fs.filer, ctx, req.Signatures)
 	}
 
 	return nil
@@ -161,119 +149,29 @@ type metadataEvent struct {
 	oldEntry     *filer.Entry
 	newEntry     *filer.Entry
 	deleteChunks bool
-	// the old entry's remote object is deleted once the rename has committed:
-	// it was copied to the new key, and a rename that fails must not lose it
-	deleteOldRemote bool
-}
-
-// pendingRemoteCopy is a remote-only rename over a remote-only target whose
-// copy runs after the commit: the store transaction cannot roll a remote
-// write back, so the new entry keeps the old key's stamp until the copy lands.
-type pendingRemoteCopy struct {
-	source  *filer.Entry
-	oldPath util.FullPath
-	newPath util.FullPath
-	failed  bool
-	copied  *filer_pb.RemoteEntry
-}
-
-// finishRemoteCopies runs the deferred remote copies after the commit. A
-// failure leaves the new entry pointing at the old key, whose object the
-// caller then keeps: remote.sync completes the copy the same way it does a
-// rename whose event it finds unfinished.
-func (fs *FilerServer) finishRemoteCopies(ctx context.Context, pendings []*pendingRemoteCopy) {
-	for _, pending := range pendings {
-		remoteEntry, err := fs.filer.CopyRemoteOnlyEntry(ctx, pending.source, pending.oldPath, pending.newPath)
-		if err != nil {
-			glog.WarningfCtx(ctx, "remote rename copy of %s: %v", pending.newPath, err)
-			pending.failed = true
-			continue
-		}
-		entry, findErr := fs.filer.FindEntry(ctx, pending.newPath)
-		if findErr != nil {
-			glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, findErr)
-			pending.failed = true
-			continue
-		}
-		entry.Remote = remoteEntry
-		delete(entry.Extended, filer.ExtRemoteRenameSourceKey)
-		if updateErr := fs.filer.UpdateEntry(ctx, entry, entry, false); updateErr != nil {
-			glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, updateErr)
-			pending.failed = true
-			continue
-		}
-		pending.copied = remoteEntry
-	}
-}
-
-// applyRemoteCopies brings the queued rename events in line with the deferred
-// copies. The stamp is written with a plain UpdateEntry, which publishes no
-// event, so the rename event carries it, without the pending-source marker,
-// to peer filers and subscribers. A failed copy keeps the old object, which
-// the entry still reads through the marker.
-func applyRemoteCopies(events []metadataEvent, pendings []*pendingRemoteCopy) {
-	for i := range events {
-		event := &events[i]
-		for _, pending := range pendings {
-			if event.oldEntry.FullPath != pending.oldPath {
-				continue
-			}
-			if pending.failed {
-				event.deleteOldRemote = false
-			} else if pending.copied != nil && event.newEntry != nil {
-				newEntry := event.newEntry.ShallowClone()
-				newEntry.Remote = pending.copied
-				newEntry.Extended = maps.Clone(newEntry.Extended)
-				delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey)
-				event.newEntry = newEntry
-			}
-		}
-	}
 }
 
 func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) {
-	if event.deleteOldRemote {
-		deleteRenamedRemoteObject(f, ctx, event.oldEntry)
-	}
 	f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures)
 }
 
-// deleteRenamedRemoteObject deletes the remote object a committed rename no
-// longer needs. A child path is not locked by its directory's rename, so
-// another rename may have put a remote-only entry at the key meanwhile; the
-// store is asked directly, a lazy fetch would revive the key. When in doubt
-// the object stays: an extra object loses nothing.
-func deleteRenamedRemoteObject(f *filer.Filer, ctx context.Context, entry *filer.Entry) {
-	current, findErr := f.Store.FindEntry(ctx, entry.FullPath)
-	switch {
-	case findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound):
-		glog.WarningfCtx(ctx, "keep remote object of renamed %s: look up its old path: %v", entry.FullPath, findErr)
-	case findErr == nil && current.IsInRemoteOnly():
-		glog.V(0).InfofCtx(ctx, "keep remote object of renamed %s: a remote-only entry has taken its path", entry.FullPath)
-	default:
-		if err := f.DeleteFromRemote(ctx, entry); err != nil {
-			glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", entry.FullPath, err)
-		}
-	}
-}
-
-func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error {
+func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error {
 
 	if err := fs.moveSelfEntry(ctx, stream, oldParent, entry, newParent, newName, func() error {
 		if entry.IsDirectory() {
-			if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures, metadataEvents, pendingChunkDeletes, pendingRemoteCopies); err != nil {
+			if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures, metadataEvents, pendingChunkDeletes); err != nil {
 				return err
 			}
 		}
 		return nil
-	}, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes, pendingRemoteCopies); err != nil {
+	}, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes); err != nil {
 		return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err)
 	}
 
 	return nil
 }
 
-func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error {
+func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error {
 
 	currentDirPath := oldParent.Child(entry.Name())
 	newDirPath := newParent.Child(newName)
@@ -296,7 +194,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb
 			// println("processing", lastFileName)
 			newChildPath := newDirPath.Child(item.Name())
 			skipTarget := fs.filer.Store.SameActualStore(newDirPath, newChildPath)
-			err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures, skipTarget, metadataEvents, pendingChunkDeletes, pendingRemoteCopies)
+			err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures, skipTarget, metadataEvents, pendingChunkDeletes)
 			if err != nil {
 				return err
 			}
@@ -308,7 +206,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb
 	return nil
 }
 
-func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error {
+func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error {
 
 	oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
 
@@ -337,48 +235,8 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		case !existingTarget.IsDirectory() && entry.IsDirectory():
 			return fmt.Errorf("%s: %w", existingTarget.FullPath, filer_pb.ErrExistingIsFile)
 		}
-	}
-
-	// a remote-only entry is copied on the remote before anything changes:
-	// deleting the old entry deletes the old object. Over a remote-only
-	// target the copy runs after the commit instead — it overwrites the
-	// target's only content, which a rolled-back commit cannot restore —
-	// and the new entry reads the old key until the copy stamps it.
-	deferredCopy := false
-	var copiedRemote *filer_pb.RemoteEntry
-	if existingTarget != nil && existingTarget.IsInRemoteOnly() && entry.IsInRemoteOnly() {
-		if _, _, copier, copyErr := fs.filer.RemoteCopyLocations(ctx, entry, oldPath, newPath); copyErr != nil {
-			return copyErr
-		} else if copier != nil {
-			deferredCopy = true
-			*pendingRemoteCopies = append(*pendingRemoteCopies, &pendingRemoteCopy{source: sourceEntry, oldPath: oldPath, newPath: newPath})
-		}
-	}
-	if !deferredCopy {
-		var copyErr error
-		copiedRemote, copyErr = fs.filer.CopyRemoteOnlyEntry(ctx, entry, oldPath, newPath)
-		if copyErr != nil {
-			return copyErr
-		}
-	}
-	remoteEntry := entry.Remote
-	if copiedRemote != nil {
-		remoteEntry = copiedRemote
-	}
-
-	if existingTarget != nil {
-		targetCtx := filer.WithSuppressedMetadataEvents(ctx)
-		if copiedRemote != nil || deferredCopy {
-			// the copy has replaced or still needs the target's object; keep it
-			targetCtx = filer.WithKeepRemoteObject(targetCtx)
-			existingTarget.Extended = maps.Clone(existingTarget.Extended)
-			if existingTarget.Extended == nil {
-				existingTarget.Extended = map[string][]byte{}
-			}
-			existingTarget.Extended[filer.ExtKeepRemoteObjectKey] = []byte("true")
-		}
 		if deleteErr := fs.filer.DeleteEntryMetaAndData(
-			targetCtx,
+			filer.WithSuppressedMetadataEvents(ctx),
 			newPath,
 			false,
 			false,
@@ -400,23 +258,9 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		Content:         entry.Content,
 		HardLinkCounter: entry.HardLinkCounter,
 		HardLinkId:      entry.HardLinkId,
-		Remote:          remoteEntry,
+		Remote:          entry.Remote,
 		Quota:           entry.Quota,
 	}
-	if deferredCopy {
-		// a pending entry renamed again still reads its original source
-		newEntry.Extended = maps.Clone(newEntry.Extended)
-		if newEntry.Extended == nil {
-			newEntry.Extended = map[string][]byte{}
-		}
-		newEntry.Extended[filer.ExtRemoteRenameSourceKey] = []byte(fs.filer.RemoteSourcePath(entry, oldPath))
-	} else if copiedRemote != nil && len(newEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
-		// the copy came from the pending source; the new key holds it now.
-		// The source itself is left to remote.sync, which deletes it only
-		// while it is still the object the rename event recorded
-		newEntry.Extended = maps.Clone(newEntry.Extended)
-		delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey)
-	}
 	if skipTargetLookup {
 		if newEntry.FullPath.IsLongerFileName(fs.filer.MaxFilenameLength) {
 			return filer_pb.ErrEntryNameTooLong
@@ -467,36 +311,19 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		})
 	}
 	*metadataEvents = append(*metadataEvents, metadataEvent{
-		oldEntry:        sourceEntry,
-		newEntry:        newEntry,
-		deleteOldRemote: copiedRemote != nil || deferredCopy,
+		oldEntry: sourceEntry,
+		newEntry: newEntry,
 	})
-	selfEvent := len(*metadataEvents) - 1
-	keepOldRemote := copiedRemote != nil || deferredCopy
 
 	if moveFolderSubEntries != nil {
 		if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
 			return moveChildrenErr
 		}
-		// removing the old directory on the remote removes every object under
-		// its prefix, the copied children's old keys included, which must wait
-		// for the commit. Those are deleted one by one after it instead, and
-		// nothing else under the prefix is touched.
-		for _, event := range (*metadataEvents)[selfEvent+1:] {
-			if event.deleteOldRemote {
-				keepOldRemote = true
-				break
-			}
-		}
 	}
 
 	// delete old entry
 	ctx = context.WithValue(ctx, "OP", "MV")
-	deleteCtx := filer.WithSuppressedMetadataEvents(ctx)
-	if keepOldRemote {
-		deleteCtx = filer.WithKeepRemoteObject(deleteCtx)
-	}
-	deleteErr := fs.filer.DeleteEntryMetaAndData(deleteCtx, oldPath, false, false, false, false, signatures, 0)
+	deleteErr := fs.filer.DeleteEntryMetaAndData(filer.WithSuppressedMetadataEvents(ctx), oldPath, false, false, false, false, signatures, 0)
 	if deleteErr != nil {
 		return deleteErr
 	}

weed/server/filer_grpc_server_rename_remote_test.go

@@ -1,430 +0,0 @@
-package weed_server
-
-import (
-	"context"
-	"errors"
-	"net/http"
-	"net/http/httptest"
-	"sync"
-	"testing"
-
-	"github.com/seaweedfs/seaweedfs/weed/filer"
-	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
-	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
-	"github.com/seaweedfs/seaweedfs/weed/util"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"google.golang.org/protobuf/proto"
-)
-
-// renameRemoteClient records the remote calls a rename makes. It cannot copy;
-// copyingRenameRemoteClient adds CopyFile.
-type renameRemoteClient struct {
-	remote_storage.RemoteStorageClient
-	mu      sync.Mutex
-	calls   []string
-	copyErr error
-}
-
-func (c *renameRemoteClient) record(call string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.calls = append(c.calls, call)
-}
-
-func (c *renameRemoteClient) recorded() []string {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	return append([]string(nil), c.calls...)
-}
-
-func (c *renameRemoteClient) StatFile(*remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) {
-	return nil, remote_storage.ErrRemoteObjectNotFound
-}
-
-func (c *renameRemoteClient) ListDirectory(context.Context, *remote_pb.RemoteStorageLocation, remote_storage.VisitFunc) error {
-	return nil
-}
-
-func (c *renameRemoteClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) error {
-	c.record("delete " + loc.Bucket + loc.Path)
-	return nil
-}
-
-func (c *renameRemoteClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) error {
-	c.record("rmdir " + loc.Bucket + loc.Path)
-	return nil
-}
-
-type copyingRenameRemoteClient struct {
-	*renameRemoteClient
-}
-
-func (c copyingRenameRemoteClient) CopyFile(src, dst *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) {
-	if c.copyErr != nil {
-		return nil, c.copyErr
-	}
-	c.record("copy " + src.Bucket + src.Path + " " + dst.Bucket + dst.Path)
-	return &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1800000000, RemoteETag: `"copy"`}, nil
-}
-
-// newRenameRemoteTestServer mounts /buckets/b, /buckets/c and /data/m on r1,
-// through client.
-func newRenameRemoteTestServer(t *testing.T, client remote_storage.RemoteStorageClient) (*FilerServer, *renameTestStore) {
-	t.Helper()
-	store := newRenameTestStore()
-	f := newRenameTestFiler(t, store)
-	f.BuildGuardedRemoteClient = func(context.Context, *remote_pb.RemoteConf, bool) (remote_storage.RemoteStorageClient, error) {
-		return client, nil
-	}
-	mapping, err := proto.Marshal(&remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{
-		"/buckets/b": {Name: "r1", Bucket: "origin", Path: "/"},
-		"/buckets/c": {Name: "r1", Bucket: "other", Path: "/"},
-		"/data/m":    {Name: "r1", Bucket: "origin", Path: "/m"},
-	}})
-	require.NoError(t, err)
-	conf, err := proto.Marshal(&remote_pb.RemoteConf{Name: "r1", Type: "s3"})
-	require.NoError(t, err)
-	for name, content := range map[string][]byte{filer.REMOTE_STORAGE_MOUNT_FILE: mapping, "r1" + filer.REMOTE_STORAGE_CONF_SUFFIX: conf} {
-		entry := newFileEntry(filer.DirectoryEtcRemote+"/"+name, 1)
-		entry.Content = content
-		store.entries[string(entry.FullPath)] = entry
-	}
-	require.NoError(t, f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f))
-	for _, dir := range []string{"/buckets", "/buckets/b", "/buckets/b/src", "/buckets/b/dst", "/buckets/c", "/data", "/data/m"} {
-		store.entries[dir] = newDirectoryEntry(dir, 2)
-	}
-	return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}, store
-}
-
-func remoteOnlyRenameEntry(path string, inode uint64) *filer.Entry {
-	entry := newFileEntry(path, inode)
-	entry.FileSize = 10
-	entry.Remote = &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1700000000, RemoteETag: `"orig"`}
-	return entry
-}
-
-func renameFile(server *FilerServer, oldPath, newPath util.FullPath) error {
-	oldDir, oldName := oldPath.DirAndName()
-	newDir, newName := newPath.DirAndName()
-	_, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{
-		OldDirectory: oldDir,
-		OldName:      oldName,
-		NewDirectory: newDir,
-		NewName:      newName,
-	})
-	return err
-}
-
-func TestRenameRemoteOnlyEntryCopiesBeforeDeletingTheOldObject(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{}}
-	server, store := newRenameRemoteTestServer(t, client)
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-
-	require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
-
-	assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded())
-	_, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg")
-	assert.ErrorIs(t, err, filer_pb.ErrNotFound)
-	dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, `"copy"`, dst.Remote.RemoteETag)
-	assert.Equal(t, uint64(101), dst.Attr.Inode)
-}
-
-func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{}}
-	server, store := newRenameRemoteTestServer(t, client)
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-	store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202)
-	queue := &captureQueue{}
-	swapNotificationQueue(t, queue)
-
-	require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
-
-	// the target's delete must not take the copy that replaced its object
-	assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded())
-	events := queue.snapshot()
-	require.Len(t, events, 2)
-	assert.True(t, filer.IsMetadataOnlyDelete(events[0].notification.OldEntry), "the target's delete event must leave the remote object alone")
-	assert.False(t, filer.IsMetadataOnlyDelete(events[1].notification.OldEntry))
-	// the copy lands after the commit; the rename event still carries its stamp
-	assert.Equal(t, `"copy"`, events[1].notification.NewEntry.RemoteEntry.GetRemoteETag())
-	dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, uint64(101), dst.Attr.Inode)
-	assert.Equal(t, `"copy"`, dst.Remote.RemoteETag)
-	assert.Empty(t, dst.Extended[filer.ExtRemoteRenameSourceKey])
-	// peers replay the post-copy state, not the source's stamp the entry
-	// committed with
-	newEvent := events[1].notification.NewEntry
-	assert.Equal(t, `"copy"`, newEvent.RemoteEntry.RemoteETag)
-	assert.Empty(t, newEvent.Extended[filer.ExtRemoteRenameSourceKey])
-}
-
-func TestRenameRemoteOnlyEntryCopyFailureKeepsTheOldObject(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}}
-	server, store := newRenameRemoteTestServer(t, client)
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-	store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202)
-
-	require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
-
-	// the copy runs after the commit over a remote-only target, so its
-	// failure cannot roll anything back: the entry still stamps the old
-	// object, which is kept for remote.sync to finish the copy
-	assert.Empty(t, client.recorded())
-	_, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg")
-	assert.ErrorIs(t, err, filer_pb.ErrNotFound)
-	dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, `"orig"`, dst.Remote.RemoteETag)
-	// reads keep resolving the source's key until the copy stamps the entry:
-	// its own key still holds the target's old object
-	assert.Equal(t, "/buckets/b/src/a.jpg", string(dst.Extended[filer.ExtRemoteRenameSourceKey]))
-	readDir, readName := server.filer.RemoteReadPath(dst, "/buckets/b/dst", "a.jpg")
-	assert.Equal(t, "/buckets/b/src", readDir)
-	assert.Equal(t, "a.jpg", readName)
-}
-
-func TestRenameRemoteOnlyEntryWithoutCopierIsRefused(t *testing.T) {
-	client := &renameRemoteClient{}
-	server, store := newRenameRemoteTestServer(t, client)
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-
-	err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")
-	require.ErrorContains(t, err, "cannot copy")
-
-	assert.Empty(t, client.recorded())
-	_, err = store.FindEntry(context.Background(), "/buckets/b/src/a.jpg")
-	require.NoError(t, err)
-	_, err = store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg")
-	assert.ErrorIs(t, err, filer_pb.ErrNotFound)
-}
-
-func TestRenameRemoteOnlyEntryOutOfItsMountIsRefused(t *testing.T) {
-	for _, paths := range [][2]util.FullPath{
-		{"/buckets/b/src/a.jpg", "/buckets/c/a.jpg"},
-		{"/data/m/a.jpg", "/data/a.jpg"},
-	} {
-		t.Run(string(paths[1]), func(t *testing.T) {
-			client := copyingRenameRemoteClient{&renameRemoteClient{}}
-			server, store := newRenameRemoteTestServer(t, client)
-			store.entries[string(paths[0])] = remoteOnlyRenameEntry(string(paths[0]), 101)
-
-			err := renameFile(server, paths[0], paths[1])
-			require.ErrorContains(t, err, "cannot be renamed out of it")
-
-			assert.Empty(t, client.recorded())
-			_, err = store.FindEntry(context.Background(), paths[0])
-			require.NoError(t, err)
-			_, err = store.FindEntry(context.Background(), paths[1])
-			assert.ErrorIs(t, err, filer_pb.ErrNotFound)
-		})
-	}
-}
-
-func TestRenameEmptyRemoteEntryDoesNotCopy(t *testing.T) {
-	// nothing to lose: no copy, and no refusal even out of the mount or
-	// without a copier; the sync writes an empty object at a new key
-	client := &renameRemoteClient{}
-	server, store := newRenameRemoteTestServer(t, client)
-	for _, path := range []string{"/buckets/b/src/empty", "/data/m/empty"} {
-		entry := remoteOnlyRenameEntry(path, 101)
-		entry.FileSize = 0
-		entry.Remote.RemoteSize = 0
-		store.entries[path] = entry
-	}
-
-	require.NoError(t, renameFile(server, "/buckets/b/src/empty", "/buckets/b/dst/empty"))
-	require.NoError(t, renameFile(server, "/data/m/empty", "/data/empty"))
-
-	assert.Equal(t, []string{"delete origin/src/empty", "delete origin/m/empty"}, client.recorded())
-}
-
-func TestRenameRemoteOnlyEntryFailingAfterTheCopyKeepsTheSource(t *testing.T) {
-	for _, src := range []util.FullPath{"/buckets/b/src/a.jpg", "/buckets/b/src"} {
-		t.Run(string(src), func(t *testing.T) {
-			client := copyingRenameRemoteClient{&renameRemoteClient{}}
-			server, store := newRenameRemoteTestServer(t, client)
-			store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-			store.commitErr = errors.New("commit failed")
-
-			err := renameFile(server, src, "/buckets/b/moved")
-			require.ErrorContains(t, err, "commit failed")
-
-			// the old object, and the old directory holding it, are deleted
-			// only once the rename has committed
-			dst := "origin/moved"
-			if src == "/buckets/b/src" {
-				dst = "origin/moved/a.jpg"
-			}
-			assert.Equal(t, []string{"copy origin/src/a.jpg " + dst}, client.recorded())
-		})
-	}
-}
-
-func TestRenameEntryWithLocalDataDoesNotCopy(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{}}
-	server, store := newRenameRemoteTestServer(t, client)
-	entry := remoteOnlyRenameEntry("/data/m/a.jpg", 101)
-	entry.Chunks = []*filer_pb.FileChunk{{FileId: "1,0123456789", Size: 10}}
-	store.entries["/data/m/a.jpg"] = entry
-
-	// even out of the mount: the old object goes, the content stays local
-	require.NoError(t, renameFile(server, "/data/m/a.jpg", "/data/a.jpg"))
-
-	assert.Equal(t, []string{"delete origin/m/a.jpg"}, client.recorded())
-	dst, err := store.FindEntry(context.Background(), "/data/a.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, `"orig"`, dst.Remote.RemoteETag)
-	assert.Len(t, dst.GetChunks(), 1)
-}
-
-func TestRenameDirectoryCopiesRemoteOnlyChildren(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{}}
-	server, store := newRenameRemoteTestServer(t, client)
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-	local := remoteOnlyRenameEntry("/buckets/b/src/b.jpg", 102)
-	local.Content = []byte("0123456789")
-	store.entries["/buckets/b/src/b.jpg"] = local
-
-	require.NoError(t, renameFile(server, "/buckets/b/src", "/buckets/b/moved"))
-
-	// b.jpg's old object goes with its entry, a.jpg's waits for the commit,
-	// and the old prefix is not swept: it could hold a newer object by then
-	assert.Equal(t, []string{
-		"copy origin/src/a.jpg origin/moved/a.jpg",
-		"delete origin/src/b.jpg",
-		"delete origin/src/a.jpg",
-	}, client.recorded())
-	moved, err := store.FindEntry(context.Background(), "/buckets/b/moved/a.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, `"copy"`, moved.Remote.RemoteETag)
-	moved, err = store.FindEntry(context.Background(), "/buckets/b/moved/b.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, `"orig"`, moved.Remote.RemoteETag)
-}
-
-func TestDeferredOldKeyDeleteKeepsAKeyRetakenSince(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{}}
-	server, store := newRenameRemoteTestServer(t, client)
-	event := metadataEvent{
-		oldEntry:        remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101),
-		newEntry:        remoteOnlyRenameEntry("/buckets/b/moved/a.jpg", 101),
-		deleteOldRemote: true,
-	}
-
-	// another rename put a remote-only entry at the old path after the commit
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 303)
-	event.notify(server.filer, context.Background(), nil)
-	assert.Empty(t, client.recorded())
-
-	// a lookup that fails says nothing about the path
-	delete(store.entries, "/buckets/b/src/a.jpg")
-	store.findErr = errors.New("store unavailable")
-	event.notify(server.filer, context.Background(), nil)
-	assert.Empty(t, client.recorded())
-
-	store.findErr = nil
-	event.notify(server.filer, context.Background(), nil)
-	assert.Equal(t, []string{"delete origin/src/a.jpg"}, client.recorded())
-}
-
-func TestRenameRemoteOnlyEntryFinishesAfterTheCommitOnASQLStore(t *testing.T) {
-	// a SQL store keeps the transaction in the context and refuses calls
-	// through it once committed, so the work after the commit must not use it
-	for _, overTarget := range []bool{false, true} {
-		t.Run(map[bool]string{false: "new path", true: "over a remote-only target"}[overTarget], func(t *testing.T) {
-			client := copyingRenameRemoteClient{&renameRemoteClient{}}
-			server, store := newRenameRemoteTestServer(t, client)
-			store.strictTx = true
-			store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-			if overTarget {
-				store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202)
-			}
-
-			require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
-
-			// the copy is stamped and the old key deleted by the filer itself
-			assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded())
-			dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg")
-			require.NoError(t, err)
-			assert.Equal(t, `"copy"`, dst.Remote.RemoteETag)
-		})
-	}
-}
-
-func TestRenamePendingEntryAgainCopiesFromItsSource(t *testing.T) {
-	client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}}
-	server, store := newRenameRemoteTestServer(t, client)
-	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
-	store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202)
-	store.entries["/buckets/b/dst/d.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/d.jpg", 303)
-
-	// the deferred copy fails: dst/a.jpg reads src/a.jpg's object
-	require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
-
-	// renamed again over another remote-only file, failing again: the marker
-	// keeps naming the original source, not dst/a.jpg's old object
-	require.NoError(t, renameFile(server, "/buckets/b/dst/a.jpg", "/buckets/b/dst/d.jpg"))
-	d, err := store.FindEntry(context.Background(), "/buckets/b/dst/d.jpg")
-	require.NoError(t, err)
-	assert.Equal(t, "/buckets/b/src/a.jpg", string(d.Extended[filer.ExtRemoteRenameSourceKey]))
-
-	// renamed to a new path, the copy comes from the source and clears the marker
-	client.copyErr = nil
-	require.NoError(t, renameFile(server, "/buckets/b/dst/d.jpg", "/buckets/b/dst/c.jpg"))
-	assert.Contains(t, client.recorded(), "copy origin/src/a.jpg origin/dst/c.jpg")
-	// the source is left to remote.sync: a concurrent rename may have copied
-	// onto that key before its entry is visible
-	assert.NotContains(t, client.recorded(), "delete origin/src/a.jpg")
-	c, err := store.FindEntry(context.Background(), "/buckets/b/dst/c.jpg")
-	require.NoError(t, err)
-	assert.Empty(t, c.Extended[filer.ExtRemoteRenameSourceKey])
-	assert.Equal(t, `"copy"`, c.Remote.RemoteETag)
-}
-
-func TestRemoteSourcePathStaysInTheMount(t *testing.T) {
-	server, _ := newRenameRemoteTestServer(t, &renameRemoteClient{})
-	entry := remoteOnlyRenameEntry("/buckets/b/x.jpg", 101)
-	for marker, want := range map[string]util.FullPath{
-		"/buckets/b/src/y.jpg": "/buckets/b/src/y.jpg", // same mount: the pending source
-		"/buckets/c/y.jpg":     "/buckets/b/x.jpg",     // another mount: ignored
-		"/tmp/y.jpg":           "/buckets/b/x.jpg",     // no mount: ignored
-	} {
-		entry.Extended = map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(marker)}
-		assert.Equal(t, want, server.filer.RemoteSourcePath(entry, "/buckets/b/x.jpg"), marker)
-	}
-}
-
-func TestTaggingCannotSetThePendingSourceMarker(t *testing.T) {
-	server, store := newRenameRemoteTestServer(t, &renameRemoteClient{})
-	store.entries["/buckets/b/x.jpg"] = remoteOnlyRenameEntry("/buckets/b/x.jpg", 101)
-
-	r := httptest.NewRequest(http.MethodPut, "/buckets/b/x.jpg?tagging", nil)
-	r.Header.Set("Seaweed-X-Remote-Rename-Source", "/buckets/b/other.jpg")
-	r.Header.Set("Seaweed-Album", "2026")
-	server.PutTaggingHandler(httptest.NewRecorder(), r)
-
-	x, err := store.FindEntry(context.Background(), "/buckets/b/x.jpg")
-	require.NoError(t, err)
-	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_grpc_server_rename_test.go

@@ -31,21 +31,6 @@ type renameTestStore struct {
 	deleteErr error
 	listErr   error         // simulates a transient store/RPC failure from a directory listing
 	findDelay time.Duration // optional: widen check-then-act windows in tests
-	findErr   error         // simulates a failing entry lookup
-	strictTx  bool          // like a SQL store: calls through a committed transaction fail
-}
-
-type renameTestTxKey struct{}
-
-type renameTestTx struct{ done bool }
-
-// txDone reports a call through a transaction that has already committed or
-// rolled back, which a SQL store refuses.
-func (s *renameTestStore) txDone(ctx context.Context) error {
-	if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok && tx.done {
-		return errors.New("sql: transaction has already been committed or rolled back")
-	}
-	return nil
 }
 
 func newRenameTestStore() *renameTestStore {
@@ -59,23 +44,10 @@ func (s *renameTestStore) GetName() string                             { return
 func (s *renameTestStore) Initialize(util.Configuration, string) error { return nil }
 func (s *renameTestStore) Shutdown()                                   {}
 func (s *renameTestStore) BeginTransaction(ctx context.Context) (context.Context, error) {
-	if s.strictTx {
-		return context.WithValue(ctx, renameTestTxKey{}, &renameTestTx{}), nil
-	}
 	return ctx, nil
 }
-func (s *renameTestStore) CommitTransaction(ctx context.Context) error {
-	if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok {
-		tx.done = true
-	}
-	return s.commitErr
-}
-func (s *renameTestStore) RollbackTransaction(ctx context.Context) error {
-	if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok {
-		tx.done = true
-	}
-	return nil
-}
+func (s *renameTestStore) CommitTransaction(context.Context) error   { return s.commitErr }
+func (s *renameTestStore) RollbackTransaction(context.Context) error { return nil }
 func (s *renameTestStore) KvPut(context.Context, []byte, []byte) error {
 	return nil
 }
@@ -84,39 +56,27 @@ func (s *renameTestStore) KvGet(context.Context, []byte) ([]byte, error) {
 }
 func (s *renameTestStore) KvDelete(context.Context, []byte) error { return nil }
 
-func (s *renameTestStore) InsertEntry(ctx context.Context, entry *filer.Entry) error {
-	if err := s.txDone(ctx); err != nil {
-		return err
-	}
+func (s *renameTestStore) InsertEntry(_ context.Context, entry *filer.Entry) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.entries[string(entry.FullPath)] = entry.ShallowClone()
 	return nil
 }
 
-func (s *renameTestStore) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
-	if err := s.txDone(ctx); err != nil {
-		return err
-	}
+func (s *renameTestStore) UpdateEntry(_ context.Context, entry *filer.Entry) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.entries[string(entry.FullPath)] = entry.ShallowClone()
 	return nil
 }
 
-func (s *renameTestStore) FindEntry(ctx context.Context, p util.FullPath) (*filer.Entry, error) {
-	if err := s.txDone(ctx); err != nil {
-		return nil, err
-	}
+func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer.Entry, error) {
 	if s.findDelay > 0 {
 		time.Sleep(s.findDelay)
 	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.findCalls[string(p)]++
-	if s.findErr != nil {
-		return nil, s.findErr
-	}
 	entry, found := s.entries[string(p)]
 	if !found {
 		return nil, filer_pb.ErrNotFound
@@ -124,10 +84,7 @@ func (s *renameTestStore) FindEntry(ctx context.Context, p util.FullPath) (*file
 	return entry.ShallowClone(), nil
 }
 
-func (s *renameTestStore) DeleteEntry(ctx context.Context, p util.FullPath) error {
-	if err := s.txDone(ctx); err != nil {
-		return err
-	}
+func (s *renameTestStore) DeleteEntry(_ context.Context, p util.FullPath) error {
 	if s.deleteErr != nil {
 		return s.deleteErr
 	}

weed/server/filer_server_handlers_read.go

@@ -206,18 +206,15 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 
 	// Every part of a multipart Range is prepared on its own, so the origin
 	// preflight below is memoized to one stat per request.
-	statOriginAt := func(readDir, readName string) func() error {
-		return sync.OnceValue(func() error {
-			client, remoteLocation, err := fs.mountedRemoteClient(ctx, readDir, readName)
-			if err != nil {
-				return err
-			}
-			_, err = client.StatFile(remoteLocation)
+	statOrigin := sync.OnceValue(func() error {
+		dir, name := entry.FullPath.DirAndName()
+		client, remoteLocation, err := fs.mountedRemoteClient(ctx, dir, name)
+		if err != nil {
 			return err
-		})
-	}
-	entryDir, entryName := entry.FullPath.DirAndName()
-	statOrigin := statOriginAt(fs.filer.RemoteReadPath(entry, entryDir, entryName))
+		}
+		_, err = client.StatFile(remoteLocation)
+		return err
+	})
 
 	ProcessRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
 		if offset+size <= int64(len(entry.Content)) {
@@ -233,23 +230,13 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		chunks := entry.GetChunks()
 		if entry.IsInRemoteOnly() {
 			dir, name := entry.FullPath.DirAndName()
-			readDir, readName := fs.filer.RemoteReadPath(entry, dir, name)
 			var mountedLocation *remote_pb.RemoteStorageLocation
 			if fs.filer.RemoteStorage != nil {
 				_, mountedLocation = fs.filer.RemoteStorage.FindMountDirectory(entry.FullPath)
 			}
 			cacheWait := remote_storage.CacheWaitTimeout(entry.Remote.RemoteSize, mountedLocation)
 			if cacheWait <= 0 {
-				streamFn, err := fs.streamFromRemoteOnly(ctx, r, readDir, readName, offset, size, statOrigin)
-				if errors.Is(err, filer_pb.ErrNotFound) && (readDir != dir || readName != name) {
-					// the pending copy finished and its source went away since
-					// the entry was read; serve what the entry names now
-					if fresh, findErr := fs.filer.FindEntry(ctx, entry.FullPath); findErr == nil {
-						freshDir, freshName := fs.filer.RemoteReadPath(fresh, dir, name)
-						return fs.streamFromRemoteOnly(ctx, r, freshDir, freshName, offset, size, statOriginAt(freshDir, freshName))
-					}
-				}
-				return streamFn, err
+				return fs.streamFromRemoteOnly(ctx, r, dir, name, offset, size, statOrigin)
 			}
 			// Bounded wait: a large download outlasts any client timeout, so
 			// serve straight from the origin once the wait expires while the
@@ -277,7 +264,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 				// when a later prepare fails; keep those on the 503 retry path.
 				if !strings.Contains(r.Header.Get("Range"), ",") {
 					glog.V(1).InfofCtx(ctx, "stream %s from remote while caching: %v", entry.FullPath, err)
-					if streamFn, remoteErr := fs.streamFromRemote(ctx, readDir, readName, offset, size); remoteErr == nil {
+					if streamFn, remoteErr := fs.streamFromRemote(ctx, dir, name, offset, size); remoteErr == nil {
 						return streamFn, nil
 					} else {
 						glog.WarningfCtx(ctx, "stream %s from remote: %v", entry.FullPath, remoteErr)

weed/server/filer_server_handlers_tagging.go

@@ -4,7 +4,6 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -36,7 +35,7 @@ func (fs *FilerServer) PutTaggingHandler(w http.ResponseWriter, r *http.Request)
 	}
 
 	for header, values := range r.Header {
-		if strings.HasPrefix(header, needle.PairNamePrefix) && header != filer.ExtRemoteRenameSourceKey {
+		if strings.HasPrefix(header, needle.PairNamePrefix) {
 			for _, value := range values {
 				existingEntry.Extended[header] = []byte(value)
 			}
@@ -90,7 +89,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/filer_server_handlers_write_autochunk.go

@@ -468,9 +468,6 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	}
 	for k, v := range r.Header {
 		if len(v) > 0 && len(v[0]) > 0 {
-			if k == filer.ExtRemoteRenameSourceKey {
-				continue // set by the filer only: it picks the remote key reads serve
-			}
 			if strings.HasPrefix(k, needle.PairNamePrefix) || k == "Cache-Control" || k == "Expires" || k == "Content-Disposition" {
 				entry.Extended[k] = []byte(v[0])
 			}

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},
 			},