admin: stop holding the shell admin lock across whole scheduler batches (#10290)

* admin: let waiting shell clients win the admin lock The admin lock manager re-acquired the cluster shell lock immediately after releasing it, while a waiting weed shell client only polls the master once a second, so an operator could lose the race indefinitely. Leave the lock free for slightly more than one poll interval before re-acquiring, and once overlapping reference-counted holds have kept the lock continuously held for over a minute, make new admin-side acquires wait for a full release before piggybacking. * admin/plugin: take the shell lock per detection and per job, not per batch The default-lane scheduler acquired the cluster shell lock once and held it across the whole pass: every due job type's detection plus all of its dispatched jobs, each drained to completion. A manual weed shell operation sharing that lock could stall behind the batch for up to the extended execution window. Hold the lock only while it protects something: around each detection scan and around each dispatched job. Manual shell operations now wait for at most one in-flight job. The admin UI detect+execute path stops wrapping dispatch in an outer hold, since a nested acquire would deadlock against the lock manager's fairness window; detection and per-job dispatch take the lock themselves. * admin/plugin: stop extending the dispatch window to the largest job estimate When any proposal's estimated runtime exceeded the remaining JobTypeMaxRuntime, the whole dispatch context was replaced with a fresh one capped at eight hours, so a single balance backlog could hold the default lane (and with it erasure_coding and vacuum detection) for that long. Keep the dispatch window at JobTypeMaxRuntime and instead detach each started attempt onto its own estimated-runtime deadline. Large jobs still get their full time once started; jobs not yet started when the window closes are canceled and re-proposed by a later detection, so sibling job types get a turn every window. * worker/balance: re-check each planned move against the master before executing A balance plan is computed at detection time, but the admin lock is released between detection and execution, so a manual shell operation can rearrange the volume in the gap. The task's own guards catch a vanished source, but a target that gained a replica in the meantime would be silently overwritten by VolumeCopy and the source delete would then reduce the volume to a single copy. Before executing each move, ask the master for the volume's current locations (uncached) and skip the move if the volume has left the source or the target already holds a replica. Skipped moves fail with a stale-move error and the next detection replans them. Without master addresses in the cluster context the check is skipped, preserving the old behavior with older admins. * admin: block re-acquire while the final lock release is in flight Release dropped the manager mutex before calling ReleaseLock, so a concurrent Acquire could see hold count zero and call RequestLock while the locker still considered itself locked. That request no-ops, leaving a hold with no live master lease. Track the in-flight release and make Acquire wait for it. Also normalize a nil release function from lock manager implementations, and make the yield and fairness windows per-instance fields so tests stop mutating globals.
2026-07-20 06:52:24 +00:00 · 2026-07-09 20:43:55 -07:00
parent 6fd82b7e6f
commit 9a7cfaf371
11 changed files with 530 additions and 86 deletions
@@ -14,16 +14,38 @@ const (
 	adminLockClientName = "admin-plugin"
 )

+const (
+	// A competing shell client polls the master once a second, while a local
+	// re-acquire would win immediately. Leave the lock free for slightly more
+	// than one poll interval so a waiting operator gets it first.
+	defaultAdminLockYieldWindow = 2 * time.Second
+	// After the lock has been held continuously for this long (overlapping
+	// reference-counted holds can keep it held indefinitely), new Acquire
+	// calls wait for a full release so shell clients are not starved.
+	defaultAdminLockFairnessWindow = time.Minute
+)
+
+// adminLocker is the subset of exclusive_locks.ExclusiveLocker the manager uses.
+type adminLocker interface {
+	RequestLock(clientName string)
+	ReleaseLock()
+	SetMessage(message string)
+}
+
 // AdminLockManager coordinates exclusive admin locks with reference counting.
 // It is safe for concurrent use.
 type AdminLockManager struct {
-	locker     *exclusive_locks.ExclusiveLocker
-	clientName string
+	locker         adminLocker
+	clientName     string
+	yieldWindow    time.Duration
+	fairnessWindow time.Duration

 	mu        sync.Mutex
 	cond      *sync.Cond
 	acquiring bool
+	releasing bool
 	holdCount int
+	heldSince time.Time

 	lastAcquiredAt time.Time
 	lastReleasedAt time.Time
@@ -40,13 +62,19 @@ func NewAdminLockManager(masterClient *wdclient.MasterClient, clientName string)
 		clientName = adminLockClientName
 	}
 	manager := &AdminLockManager{
-		locker:     exclusive_locks.NewExclusiveLocker(masterClient, adminLockName),
-		clientName: clientName,
+		locker:         exclusive_locks.NewExclusiveLocker(masterClient, adminLockName),
+		clientName:     clientName,
+		yieldWindow:    defaultAdminLockYieldWindow,
+		fairnessWindow: defaultAdminLockFairnessWindow,
 	}
 	manager.cond = sync.NewCond(&manager.mu)
 	return manager
 }

+// Acquire takes the shared cluster lock, reference counted so overlapping
+// admin-side holders piggyback on one lease. It must not be called while the
+// caller's own call stack already holds the lock: once the fairness window
+// makes new acquires wait for a full release, a nested Acquire deadlocks.
 func (m *AdminLockManager) Acquire(reason string) (func(), error) {
 	if m == nil || m.locker == nil {
 		return func() {}, nil
@@ -57,18 +85,28 @@ func (m *AdminLockManager) Acquire(reason string) (func(), error) {
 		m.locker.SetMessage(reason)
 		m.currentReason = reason
 	}
-	for m.acquiring {
+	for m.acquiring || m.releasing || (m.holdCount > 0 && time.Since(m.heldSince) > m.fairnessWindow) {
 		m.cond.Wait()
 	}
 	if m.holdCount == 0 {
 		m.acquiring = true
 		m.waitingSince = time.Now().UTC()
 		m.waitingReason = reason
+		yield := time.Duration(0)
+		if !m.lastReleasedAt.IsZero() {
+			if since := time.Since(m.lastReleasedAt); since < m.yieldWindow {
+				yield = m.yieldWindow - since
+			}
+		}
 		m.mu.Unlock()
+		if yield > 0 {
+			time.Sleep(yield)
+		}
 		m.locker.RequestLock(m.clientName)
 		m.mu.Lock()
 		m.acquiring = false
 		m.holdCount = 1
+		m.heldSince = time.Now()
 		m.lastAcquiredAt = time.Now().UTC()
 		m.waitingSince = time.Time{}
 		m.waitingReason = ""
@@ -96,14 +134,22 @@ func (m *AdminLockManager) Release() {
 	}
 	m.holdCount--
 	shouldRelease := m.holdCount == 0
+	if shouldRelease {
+		// Block new acquires until ReleaseLock completes: a RequestLock
+		// issued while the locker still considers itself locked would no-op
+		// and leave a hold with no live lease.
+		m.releasing = true
+	}
 	m.mu.Unlock()

 	if shouldRelease {
+		m.locker.ReleaseLock()
 		m.mu.Lock()
+		m.releasing = false
 		m.lastReleasedAt = time.Now().UTC()
 		m.currentReason = ""
+		m.cond.Broadcast()
 		m.mu.Unlock()
-		m.locker.ReleaseLock()
 	}
 }

@@ -0,0 +1,156 @@
+package dash
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+type fakeAdminLocker struct {
+	releaseDelay time.Duration
+
+	mu       sync.Mutex
+	requests []time.Time
+	releases []time.Time // recorded when ReleaseLock returns
+}
+
+func (f *fakeAdminLocker) RequestLock(clientName string) {
+	f.mu.Lock()
+	f.requests = append(f.requests, time.Now())
+	f.mu.Unlock()
+}
+
+func (f *fakeAdminLocker) ReleaseLock() {
+	if f.releaseDelay > 0 {
+		time.Sleep(f.releaseDelay)
+	}
+	f.mu.Lock()
+	f.releases = append(f.releases, time.Now())
+	f.mu.Unlock()
+}
+
+func (f *fakeAdminLocker) SetMessage(message string) {}
+
+func (f *fakeAdminLocker) counts() (int, int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.requests), len(f.releases)
+}
+
+func newTestLockManager(locker adminLocker, yieldWindow, fairnessWindow time.Duration) *AdminLockManager {
+	m := &AdminLockManager{
+		locker:         locker,
+		clientName:     "test",
+		yieldWindow:    yieldWindow,
+		fairnessWindow: fairnessWindow,
+	}
+	m.cond = sync.NewCond(&m.mu)
+	return m
+}
+
+func TestAdminLockManagerRefCountsOverlappingHolds(t *testing.T) {
+	t.Parallel()
+
+	locker := &fakeAdminLocker{}
+	m := newTestLockManager(locker, defaultAdminLockYieldWindow, defaultAdminLockFairnessWindow)
+
+	release1, _ := m.Acquire("a")
+	release2, _ := m.Acquire("b")
+
+	if requests, releases := locker.counts(); requests != 1 || releases != 0 {
+		t.Fatalf("expected one lease request while overlapping, got requests=%d releases=%d", requests, releases)
+	}
+
+	release1()
+	if _, releases := locker.counts(); releases != 0 {
+		t.Fatalf("lock released while still held by another caller")
+	}
+	release2()
+	if _, releases := locker.counts(); releases != 1 {
+		t.Fatalf("expected one release after last holder, got %d", releases)
+	}
+}
+
+func TestAdminLockManagerYieldsBeforeReacquire(t *testing.T) {
+	t.Parallel()
+
+	locker := &fakeAdminLocker{}
+	m := newTestLockManager(locker, 100*time.Millisecond, defaultAdminLockFairnessWindow)
+
+	release, _ := m.Acquire("first")
+	release()
+	release, _ = m.Acquire("second")
+	release()
+
+	locker.mu.Lock()
+	gap := locker.requests[1].Sub(locker.releases[0])
+	locker.mu.Unlock()
+	if gap < 90*time.Millisecond {
+		t.Fatalf("re-acquire did not yield to competing clients: gap=%v", gap)
+	}
+}
+
+func TestAdminLockManagerFairnessWindowForcesFullRelease(t *testing.T) {
+	t.Parallel()
+
+	locker := &fakeAdminLocker{}
+	m := newTestLockManager(locker, 20*time.Millisecond, 50*time.Millisecond)
+
+	release1, _ := m.Acquire("long-hold")
+	m.mu.Lock()
+	m.heldSince = time.Now().Add(-time.Second)
+	m.mu.Unlock()
+
+	acquired := make(chan func(), 1)
+	go func() {
+		release2, _ := m.Acquire("late-joiner")
+		acquired <- release2
+	}()
+
+	select {
+	case <-acquired:
+		t.Fatalf("late acquire joined a hold older than the fairness window")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	release1()
+
+	select {
+	case release2 := <-acquired:
+		release2()
+	case <-time.After(time.Second):
+		t.Fatalf("late acquire did not proceed after full release")
+	}
+
+	if requests, releases := locker.counts(); requests != 2 || releases != 2 {
+		t.Fatalf("expected a full release/re-acquire cycle, got requests=%d releases=%d", requests, releases)
+	}
+}
+
+func TestAdminLockManagerAcquireWaitsForReleaseInFlight(t *testing.T) {
+	t.Parallel()
+
+	locker := &fakeAdminLocker{releaseDelay: 50 * time.Millisecond}
+	m := newTestLockManager(locker, time.Millisecond, defaultAdminLockFairnessWindow)
+
+	release1, _ := m.Acquire("first")
+	releaseDone := make(chan struct{})
+	go func() {
+		release1()
+		close(releaseDone)
+	}()
+	// Let the release enter the slow ReleaseLock before re-acquiring.
+	time.Sleep(10 * time.Millisecond)
+	release2, _ := m.Acquire("second")
+	<-releaseDone
+	release2()
+
+	locker.mu.Lock()
+	defer locker.mu.Unlock()
+	if len(locker.requests) != 2 || len(locker.releases) != 2 {
+		t.Fatalf("unexpected lease traffic: requests=%d releases=%d", len(locker.requests), len(locker.releases))
+	}
+	if locker.requests[1].Before(locker.releases[0]) {
+		t.Fatalf("RequestLock ran while ReleaseLock was still in flight")
+	}
+}
@@ -1459,8 +1459,8 @@ func (s *AdminServer) RunPluginDetectionWithReport(

 // DispatchPluginProposals dispatches a batch of proposals using the same
 // capacity-aware dispatch logic as the scheduler loop (executor reservation with
-// backoff, per-job retry on transient errors). The plugin lock must already be
-// held by the caller.
+// backoff, per-job retry on transient errors). The dispatch takes the cluster
+// admin lock around each job itself; callers must not hold it.
 func (s *AdminServer) DispatchPluginProposals(
 	ctx context.Context,
 	jobType string,
@@ -595,20 +595,15 @@ func (s *AdminServer) TriggerPluginDetectionAPI(w http.ResponseWriter, r *http.R
 }

 // RunPluginJobTypeAPI runs full workflow for one job type: detect then dispatch detected jobs.
+// RunPluginDetection and the dispatch path take the cluster admin lock
+// themselves (around detection and around each job), so no lock is held
+// across the whole workflow here.
 func (s *AdminServer) RunPluginJobTypeAPI(w http.ResponseWriter, r *http.Request) {
 	jobType := strings.TrimSpace(mux.Vars(r)["jobType"])
 	if jobType == "" {
 		writeJSONError(w, http.StatusBadRequest, "jobType is required")
 		return
 	}
-	releaseLock, err := s.acquirePluginLock(fmt.Sprintf("plugin detect+execute %s", jobType))
-	if err != nil {
-		writeJSONError(w, http.StatusInternalServerError, err.Error())
-		return
-	}
-	if releaseLock != nil {
-		defer releaseLock()
-	}

 	var req struct {
 		ClusterContext json.RawMessage `json:"cluster_context"`
@@ -424,7 +424,13 @@ func (r *Plugin) acquireAdminLock(reason string) (func(), error) {
 	if r == nil || r.lockManager == nil {
 		return func() {}, nil
 	}
-	return r.lockManager.Acquire(reason)
+	release, err := r.lockManager.Acquire(reason)
+	if release == nil {
+		// Lock manager implementations may return a nil release when
+		// disabled; callers always invoke the release function.
+		release = func() {}
+	}
+	return release, err
 }

 // RunDetectionWithReport requests one detector worker and returns proposals with request metadata.
@@ -98,9 +98,11 @@ func (r *Plugin) laneSchedulerLoop(ls *schedulerLaneState) {
 // runLaneSchedulerIteration runs one scheduling pass for a single lane,
 // processing only the job types assigned to that lane.
 //
-// For lanes that require a lock (e.g. LaneDefault), all job types are
-// processed sequentially under one admin lock because their volume
-// management operations share global state.
+// For lanes that require a lock (e.g. LaneDefault), job types are processed
+// sequentially because their volume management operations share global
+// state, and the cluster admin lock is taken around each detection and each
+// dispatched job rather than the whole pass, so manual shell operations
+// sharing the same lock can interleave between jobs.
 //
 // For lanes that do not require a lock (e.g. LaneIceberg, LaneLifecycle),
 // each job type runs independently in its own goroutine so they do not
@@ -122,7 +124,7 @@ func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool {
 	}

 	if LaneRequiresLock(ls.lane) {
-		return r.runLaneSchedulerIterationLocked(ls, jobTypes)
+		return r.runLaneSchedulerIterationSequential(ls, jobTypes)
 	}
 	return r.runLaneSchedulerIterationConcurrent(ls, jobTypes)
 }
@@ -162,21 +164,13 @@ func (r *Plugin) collectDueJobTypes(ls *schedulerLaneState, jobTypes []string) (
 	return active, due
 }

-// runLaneSchedulerIterationLocked processes job types sequentially under a
-// single admin lock. Used by the default lane where volume management
-// operations must be serialised.
-func (r *Plugin) runLaneSchedulerIterationLocked(ls *schedulerLaneState, jobTypes []string) bool {
-	r.setLaneLoopState(ls, "", "waiting_for_lock")
-	lockName := fmt.Sprintf("plugin scheduler:%s", ls.lane)
-	releaseLock, err := r.acquireAdminLock(lockName)
-	if err != nil {
-		glog.Warningf("Plugin scheduler [%s] failed to acquire lock: %v", ls.lane, err)
-		r.setLaneLoopState(ls, "", "idle")
-		return false
-	}
-	if releaseLock != nil {
-		defer releaseLock()
-	}
+// runLaneSchedulerIterationSequential processes job types one at a time.
+// Used by the default lane where volume management operations must be
+// serialised. The cluster admin lock is not held across the pass;
+// runJobTypeIteration and dispatchScheduledProposals take it around each
+// detection and each job.
+func (r *Plugin) runLaneSchedulerIterationSequential(ls *schedulerLaneState, jobTypes []string) bool {
+	r.setLaneLoopState(ls, "", "busy")

 	active, due := r.collectDueJobTypes(ls, jobTypes)
 	hadJobs := false
@@ -288,6 +282,30 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 		return false
 	}

+	// Detection plans against the live topology, so it runs under the shared
+	// cluster admin lock; the lock is released as soon as detection returns.
+	releaseDetectionLock := func() {}
+	if LaneRequiresLock(JobTypeLane(jobType)) {
+		r.setSchedulerLoopStateForJobType(jobType, "waiting_for_lock")
+		release, lockErr := r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s detection", jobType))
+		if lockErr != nil {
+			r.recordSchedulerDetectionError(jobType, lockErr)
+			r.appendActivity(JobActivity{
+				JobType:    jobType,
+				Source:     "admin_scheduler",
+				Message:    fmt.Sprintf("scheduled detection aborted: %v", lockErr),
+				Stage:      "failed",
+				OccurredAt: timeToPtr(time.Now().UTC()),
+			})
+			r.recordSchedulerRunComplete(jobType, "error")
+			return false
+		}
+		var releaseOnce sync.Once
+		releaseDetectionLock = func() { releaseOnce.Do(release) }
+		r.setSchedulerLoopStateForJobType(jobType, "detecting")
+	}
+	defer releaseDetectionLock()
+
 	detectionTimeout := policy.DetectionTimeout
 	remaining := time.Until(start.Add(maxRuntime))
 	if remaining <= 0 {
@@ -311,6 +329,7 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 	detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout)
 	proposals, err := r.RunDetection(detectCtx, jobType, clusterContext, policy.MaxResults)
 	cancelDetect()
+	releaseDetectionLock()
 	if err != nil {
 		r.recordSchedulerDetectionError(jobType, err)
 		stage := "failed"
@@ -375,26 +394,6 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo

 	r.setSchedulerLoopStateForJobType(jobType, "executing")

-	// Scan proposals for the maximum estimated_runtime_seconds so the
-	// execution phase gets enough time for large jobs (e.g. vacuum on
-	// big volumes). If any proposal needs more time than the remaining
-	// JobTypeMaxRuntime, extend the execution context accordingly.
-	var maxEstimatedRuntime time.Duration
-	for _, p := range filtered {
-		if p.Parameters != nil {
-			if est, ok := p.Parameters["estimated_runtime_seconds"]; ok {
-				if v := est.GetInt64Value(); v > 0 {
-					if d := time.Duration(v) * time.Second; d > maxEstimatedRuntime {
-						maxEstimatedRuntime = d
-					}
-				}
-			}
-		}
-	}
-	if maxEstimatedRuntime > maxEstimatedRuntimeCap {
-		maxEstimatedRuntime = maxEstimatedRuntimeCap
-	}
-
 	remaining = time.Until(start.Add(maxRuntime))
 	if remaining <= 0 {
 		r.appendActivity(JobActivity{
@@ -408,17 +407,6 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 		return detected
 	}

-	// If the longest estimated job exceeds the remaining JobTypeMaxRuntime,
-	// create a new execution context with enough headroom instead of using
-	// jobCtx which would cancel too early.
-	execCtx := jobCtx
-	execCancel := context.CancelFunc(func() {})
-	if maxEstimatedRuntime > 0 && maxEstimatedRuntime > remaining {
-		execCtx, execCancel = context.WithTimeout(context.Background(), maxEstimatedRuntime)
-		remaining = maxEstimatedRuntime
-	}
-	defer execCancel()
-
 	execPolicy := policy
 	if execPolicy.ExecutionTimeout <= 0 {
 		execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout
@@ -427,10 +415,15 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
 		execPolicy.ExecutionTimeout = remaining
 	}

-	successCount, errorCount, canceledCount := r.dispatchScheduledProposals(execCtx, jobType, filtered, clusterContext, execPolicy)
+	// jobCtx bounds how long this run keeps STARTING jobs; a started job
+	// runs on its own estimated-runtime deadline (see
+	// executeScheduledJobWithExecutor). Jobs still queued when the window
+	// closes are canceled and re-proposed by a later detection, so one large
+	// backlog cannot monopolise the lane for hours.
+	successCount, errorCount, canceledCount := r.dispatchScheduledProposals(jobCtx, jobType, filtered, clusterContext, execPolicy)

 	status := "success"
-	if execCtx.Err() != nil {
+	if jobCtx.Err() != nil {
 		status = "timeout"
 	} else if errorCount > 0 || canceledCount > 0 {
 		status = "error"
@@ -818,6 +811,11 @@ func (r *Plugin) dispatchScheduledProposals(
 		ctx = context.Background()
 	}

+	// Volume management jobs mutate volume placement, so each one runs under
+	// the shared cluster admin lock. Taking it per job instead of around the
+	// whole dispatch lets manual shell operations interleave between jobs.
+	jobsRequireLock := LaneRequiresLock(JobTypeLane(jobType))
+
 	jobQueue := make(chan *plugin_pb.JobSpec, len(proposals))
 	for index, proposal := range proposals {
 		job := buildScheduledJobSpec(jobType, proposal, index)
@@ -900,7 +898,42 @@ func (r *Plugin) dispatchScheduledProposals(
 						break
 					}

+					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()
 					if errors.Is(err, errExecutorAtCapacity) {
 						r.trackExecutionQueued(job)
@@ -1164,10 +1197,6 @@ func (r *Plugin) executeScheduledJobWithExecutor(
 			return ctx.Err()
 		}

-		parent := ctx
-		if parent == nil {
-			parent = context.Background()
-		}
 		// Use the job's estimated runtime if provided and larger than the
 		// default execution timeout. This lets handlers like vacuum scale
 		// the timeout based on volume size so large volumes are not killed.
@@ -1185,7 +1214,11 @@ func (r *Plugin) executeScheduledJobWithExecutor(
 				}
 			}
 		}
-		execCtx, cancel := context.WithTimeout(parent, timeout)
+		// The attempt runs against its own deadline, detached from the
+		// dispatch window (ctx): a job started near the end of the window
+		// may run to completion; the window only stops further jobs from
+		// starting.
+		execCtx, cancel := context.WithTimeout(context.Background(), timeout)
 		_, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt))
 		cancel()
 		if err == nil {
@@ -664,6 +664,12 @@ func TestRunLaneSchedulerIterationLockBehavior(t *testing.T) {
 				t.Fatalf("SaveJobTypeConfig: %v", err)
 			}

+			// Make the job type due immediately so the iteration reaches
+			// detection; the lock is only taken around due work now.
+			pluginSvc.schedulerMu.Lock()
+			pluginSvc.nextDetectionAt[tt.jobType] = time.Now().UTC().Add(-time.Second)
+			pluginSvc.schedulerMu.Unlock()
+
 			ls := pluginSvc.lanes[tt.lane]
 			pluginSvc.runLaneSchedulerIteration(ls)

@@ -674,6 +680,61 @@ func TestRunLaneSchedulerIterationLockBehavior(t *testing.T) {
 	}
 }

+func TestDispatchScheduledProposalsLocksPerJob(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name      string
+		jobType   string
+		wantLocks int
+	}{
+		{"DefaultLaneLocksEachJob", "volume_balance", 3},
+		{"LifecycleLaneNeedsNoLock", "s3_lifecycle", 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			lm := &trackingLockManager{}
+			pluginSvc, err := New(Options{LockManager: lm})
+			if err != nil {
+				t.Fatalf("New: %v", err)
+			}
+			defer pluginSvc.Shutdown()
+
+			// The worker has capabilities but no connected session, so each
+			// job reserves capacity, fails to send, and moves on.
+			pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+				WorkerId: "worker-a",
+				Capabilities: []*plugin_pb.JobTypeCapability{
+					{JobType: tt.jobType, CanExecute: true, MaxExecutionConcurrency: 4},
+				},
+			})
+
+			policy := schedulerPolicy{
+				ExecutionConcurrency:   1,
+				PerWorkerConcurrency:   1,
+				ExecutionTimeout:       time.Second,
+				ExecutorReserveBackoff: time.Millisecond,
+			}
+			proposals := []*plugin_pb.JobProposal{
+				{ProposalId: "p1", JobType: tt.jobType, DedupeKey: "k1"},
+				{ProposalId: "p2", JobType: tt.jobType, DedupeKey: "k2"},
+				{ProposalId: "p3", JobType: tt.jobType, DedupeKey: "k3"},
+			}
+
+			success, errors, canceled := pluginSvc.dispatchScheduledProposals(
+				context.Background(), tt.jobType, proposals, &plugin_pb.ClusterContext{}, policy)
+			if success != 0 || canceled != 0 || errors != 3 {
+				t.Fatalf("unexpected dispatch counts: success=%d errors=%d canceled=%d", success, errors, canceled)
+			}
+			if got := lm.count(); got != tt.wantLocks {
+				t.Fatalf("lock acquired %d times, want %d", got, tt.wantLocks)
+			}
+		})
+	}
+}
+
 // ---------- lane-scoped prune ----------

 func TestPruneSchedulerState_DefaultLaneKeepsForeignLanesAndPrunesOwnStale(t *testing.T) {
@@ -6,8 +6,8 @@ import (
 )

 // SchedulerLane identifies an independent scheduling track. Each lane runs
-// its own goroutine, maintains its own detection timing, and acquires its
-// own admin lock so that workloads in different lanes never block each other.
+// its own goroutine and maintains its own detection timing so that
+// workloads in different lanes never block each other.
 type SchedulerLane string

 const (
@@ -39,17 +39,19 @@ var laneIdleSleep = map[SchedulerLane]time.Duration{
 }

 // laneRequiresLock maps each lane to whether its job types must be
-// serialised under a single admin lock. The default lane needs this
-// because volume-management operations share global state. Other
-// lanes run each job type independently.
+// serialised and run under the shared cluster admin lock. The default
+// lane needs this because volume-management operations share global
+// state; the lock is taken around each detection and each job so manual
+// shell operations can interleave. Other lanes run each job type
+// independently without the lock.
 var laneRequiresLock = map[SchedulerLane]bool{
 	LaneDefault:   true,
 	LaneIceberg:   false,
 	LaneLifecycle: false,
 }

-// LaneRequiresLock returns true if the given lane needs a single admin
-// lock to serialise its job types. Unknown lanes default to true.
+// LaneRequiresLock returns true if the given lane serialises its job types
+// and runs them under the cluster admin lock. Unknown lanes default to true.
 func LaneRequiresLock(lane SchedulerLane) bool {
 	if v, ok := laneRequiresLock[lane]; ok {
 		return v
@@ -0,0 +1,75 @@
+package pluginworker
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+)
+
+// LookupVolumeLocations asks the masters for the current registered locations
+// of a volume, bypassing any client-side caches. Execution handlers use it to
+// re-check a proposal against the live topology, since a proposal can go
+// stale between detection and execution. Returns the location URLs
+// (host:port) the master reports; an empty slice means the volume is not
+// registered anywhere.
+func LookupVolumeLocations(ctx context.Context, masterAddresses []string, grpcDialOption grpc.DialOption, volumeID uint32) ([]string, error) {
+	if grpcDialOption == nil {
+		return nil, fmt.Errorf("grpc dial option is not configured")
+	}
+	if len(masterAddresses) == 0 {
+		return nil, fmt.Errorf("no master addresses provided in cluster context")
+	}
+
+	vid := strconv.FormatUint(uint64(volumeID), 10)
+	var lastErr error
+	for _, address := range masterAddresses {
+		for _, candidate := range MasterAddressCandidates(address) {
+			if ctx.Err() != nil {
+				return nil, ctx.Err()
+			}
+
+			dialCtx, cancelDial := context.WithTimeout(ctx, 5*time.Second)
+			conn, err := pb.GrpcDial(dialCtx, candidate, false, grpcDialOption)
+			cancelDial()
+			if err != nil {
+				lastErr = err
+				continue
+			}
+
+			client := master_pb.NewSeaweedClient(conn)
+			callCtx, cancelCall := context.WithTimeout(ctx, 10*time.Second)
+			resp, callErr := client.LookupVolume(callCtx, &master_pb.LookupVolumeRequest{
+				VolumeOrFileIds: []string{vid},
+			})
+			cancelCall()
+			_ = conn.Close()
+			if callErr != nil {
+				lastErr = callErr
+				continue
+			}
+
+			for _, vidLocation := range resp.VolumeIdLocations {
+				if vidLocation.VolumeOrFileId != vid {
+					continue
+				}
+				locations := make([]string, 0, len(vidLocation.Locations))
+				for _, location := range vidLocation.Locations {
+					locations = append(locations, location.Url)
+				}
+				return locations, nil
+			}
+			// The master answered but does not know the volume.
+			return nil, nil
+		}
+	}
+
+	if lastErr == nil {
+		lastErr = fmt.Errorf("no valid master address candidate")
+	}
+	return nil, lastErr
+}
@@ -723,19 +723,23 @@ func (h *VolumeBalanceHandler) executeSingleMove(
 		return err
 	}

-	if err := task.Execute(execCtx, params); err != nil {
+	execErr := h.checkMoveStillValid(execCtx, request.GetClusterContext().GetMasterGrpcAddresses(), params.VolumeId, params.Sources[0].Node, params.Targets[0].Node)
+	if execErr == nil {
+		execErr = task.Execute(execCtx, params)
+	}
+	if execErr != nil {
 		_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
 			JobId:           request.Job.JobId,
 			JobType:         request.Job.JobType,
 			State:           plugin_pb.JobState_JOB_STATE_FAILED,
 			ProgressPercent: 100,
 			Stage:           "failed",
-			Message:         err.Error(),
+			Message:         execErr.Error(),
 			Activities: []*plugin_pb.ActivityEvent{
-				pluginworker.BuildExecutorActivity("failed", err.Error()),
+				pluginworker.BuildExecutorActivity("failed", execErr.Error()),
 			},
 		})
-		return err
+		return execErr
 	}

 	sourceNode := params.Sources[0].Node
@@ -766,6 +770,41 @@ func (h *VolumeBalanceHandler) executeSingleMove(
 	})
 }

+// checkMoveStillValid re-checks a planned move against the master's live
+// view right before executing it. The admin lock is released between
+// detection and execution, so the plan can go stale: the volume may have
+// left the source, or the target may have gained a replica that the copy
+// would overwrite and the source delete would then reduce to a single copy.
+// Without master addresses (older admin) the check is skipped and the move
+// relies on the task's own execution-time guards.
+func (h *VolumeBalanceHandler) checkMoveStillValid(ctx context.Context, masterAddresses []string, volumeID uint32, sourceNode, targetNode string) error {
+	if len(masterAddresses) == 0 {
+		glog.Warningf("volume balance: no master addresses in cluster context, skipping pre-move check for volume %d", volumeID)
+		return nil
+	}
+	locations, err := pluginworker.LookupVolumeLocations(ctx, masterAddresses, h.grpcDialOption, volumeID)
+	if err != nil {
+		return fmt.Errorf("pre-move check for volume %d: %w", volumeID, err)
+	}
+	return checkMovePreconditions(locations, volumeID, sourceNode, targetNode)
+}
+
+func checkMovePreconditions(locations []string, volumeID uint32, sourceNode, targetNode string) error {
+	sourceFound := false
+	for _, location := range locations {
+		switch strings.TrimSpace(location) {
+		case targetNode:
+			return fmt.Errorf("stale move: volume %d already has a replica on target %s", volumeID, targetNode)
+		case sourceNode:
+			sourceFound = true
+		}
+	}
+	if !sourceFound {
+		return fmt.Errorf("stale move: volume %d is no longer on source %s", volumeID, sourceNode)
+	}
+	return nil
+}
+
 // executeBatchMoves runs multiple volume moves concurrently within a single job.
 func (h *VolumeBalanceHandler) executeBatchMoves(
 	ctx context.Context,
@@ -881,6 +920,7 @@ func (h *VolumeBalanceHandler) executeBatchMoves(

 	sem := make(chan struct{}, maxConcurrent)
 	results := make(chan moveResult, totalMoves)
+	masterAddresses := request.GetClusterContext().GetMasterGrpcAddresses()

 	for i, move := range moves {
 		sem <- struct{}{} // acquire slot
@@ -899,7 +939,10 @@ func (h *VolumeBalanceHandler) executeBatchMoves(
 			})

 			moveParams := buildMoveTaskParams(m, bp)
-			err := task.Execute(batchCtx, moveParams)
+			err := h.checkMoveStillValid(batchCtx, masterAddresses, m.VolumeId, m.SourceNode, m.TargetNode)
+			if err == nil {
+				err = task.Execute(batchCtx, moveParams)
+			}
 			results <- moveResult{
 				index:    idx,
 				volumeID: m.VolumeId,
@@ -795,3 +795,30 @@ func (r *recordingDetectionSender) SendActivity(event *plugin_pb.ActivityEvent)
 	}
 	return nil
 }
+
+func TestCheckMovePreconditions(t *testing.T) {
+	tests := []struct {
+		name      string
+		locations []string
+		wantErr   string
+	}{
+		{"valid move", []string{"10.0.0.1:8080", "10.0.0.3:8080"}, ""},
+		{"volume left the source", []string{"10.0.0.3:8080"}, "no longer on source"},
+		{"volume gone entirely", nil, "no longer on source"},
+		{"target already has a replica", []string{"10.0.0.1:8080", "10.0.0.2:8080"}, "already has a replica on target"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := checkMovePreconditions(tt.locations, 5, "10.0.0.1:8080", "10.0.0.2:8080")
+			if tt.wantErr == "" {
+				if err != nil {
+					t.Fatalf("unexpected error: %v", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("expected error containing %q, got %v", tt.wantErr, err)
+			}
+		})
+	}
+}