admin: make the cluster lock's fairness window configurable (#12426)

* admin: make the cluster lock's fairness window configurable

Scheduled jobs hold the cluster admin lock for their whole run. Once the
lock had been held for a minute, every new job waited until all running
jobs had released it, so with hour-long EC jobs dispatch ran in waves and
most workers sat idle between them.

-adminLock.fairnessWindow keeps the one-minute default; 0 lets new jobs
join the shared hold, and shell clients get the lock when the scheduler
next drains.

Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi

* admin: wait for the blocked acquire before disabling the fairness window

The wake-up test slept 50 ms before SetFairnessWindow(0), so a late
acquire that started after the change joined without exercising the
Broadcast. Wait until it is registered as a waiter instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/admin/dash/admin_lock_manager.go

@@ -23,7 +23,7 @@ const (
 	// 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
+	DefaultAdminLockFairnessWindow = time.Minute
 )
 
 // adminLocker is the subset of exclusive_locks.ExclusiveLocker the manager uses.
@@ -67,7 +67,7 @@ func NewAdminLockManager(masterClient *wdclient.MasterClient, clientName string)
 		locker:         exclusive_locks.NewExclusiveLocker(masterClient, adminLockName),
 		clientName:     clientName,
 		yieldWindow:    defaultAdminLockYieldWindow,
-		fairnessWindow: defaultAdminLockFairnessWindow,
+		fairnessWindow: DefaultAdminLockFairnessWindow,
 	}
 	manager.cond = sync.NewCond(&manager.mu)
 	return manager
@@ -93,7 +93,7 @@ func (m *AdminLockManager) Acquire(reason string) (func(), error) {
 		delete(m.waiters, waiter)
 		m.mu.Unlock()
 	}()
-	for m.acquiring || m.releasing || (m.holdCount > 0 && time.Since(m.heldSince) > m.fairnessWindow) {
+	for m.acquiring || m.releasing || m.fairnessDueLocked() {
 		m.cond.Wait()
 	}
 	if reason != "" {
@@ -134,6 +134,26 @@ func (m *AdminLockManager) Acquire(reason string) (func(), error) {
 	return m.Release, nil
 }
 
+// fairnessDueLocked reports whether new holders must wait for a full release
+// because the lock has been held past the fairness window. A window of zero
+// or less never forces one. Caller holds m.mu.
+func (m *AdminLockManager) fairnessDueLocked() bool {
+	return m.fairnessWindow > 0 && m.holdCount > 0 && time.Since(m.heldSince) > m.fairnessWindow
+}
+
+// SetFairnessWindow changes how long overlapping holds may keep the lock
+// before new Acquire calls wait for a full release; zero or less disables
+// the forced release, leaving shell clients the gaps between holds.
+func (m *AdminLockManager) SetFairnessWindow(window time.Duration) {
+	if m == nil {
+		return
+	}
+	m.mu.Lock()
+	m.fairnessWindow = window
+	m.cond.Broadcast()
+	m.mu.Unlock()
+}
+
 func (m *AdminLockManager) Release() {
 	if m == nil || m.locker == nil {
 		return

weed/admin/dash/admin_lock_manager_test.go

@@ -89,7 +89,7 @@ func TestAdminLockManagerRefCountsOverlappingHolds(t *testing.T) {
 	t.Parallel()
 
 	locker := &fakeAdminLocker{}
-	m := newTestLockManager(locker, defaultAdminLockYieldWindow, defaultAdminLockFairnessWindow)
+	m := newTestLockManager(locker, defaultAdminLockYieldWindow, DefaultAdminLockFairnessWindow)
 
 	release1, _ := m.Acquire("a")
 	release2, _ := m.Acquire("b")
@@ -112,7 +112,7 @@ func TestAdminLockManagerYieldsBeforeReacquire(t *testing.T) {
 	t.Parallel()
 
 	locker := &fakeAdminLocker{}
-	m := newTestLockManager(locker, 100*time.Millisecond, defaultAdminLockFairnessWindow)
+	m := newTestLockManager(locker, 100*time.Millisecond, DefaultAdminLockFairnessWindow)
 
 	release, _ := m.Acquire("first")
 	release()
@@ -178,7 +178,7 @@ func TestAdminLockManagerAcquireWaitsForReleaseInFlight(t *testing.T) {
 	t.Parallel()
 
 	locker := &fakeAdminLocker{releaseDelay: 50 * time.Millisecond}
-	m := newTestLockManager(locker, time.Millisecond, defaultAdminLockFairnessWindow)
+	m := newTestLockManager(locker, time.Millisecond, DefaultAdminLockFairnessWindow)
 
 	release1, _ := m.Acquire("first")
 	releaseDone := make(chan struct{})
@@ -201,3 +201,72 @@ func TestAdminLockManagerAcquireWaitsForReleaseInFlight(t *testing.T) {
 		t.Fatalf("RequestLock ran while ReleaseLock was still in flight")
 	}
 }
+
+func TestAdminLockManagerZeroFairnessWindowKeepsJoining(t *testing.T) {
+	t.Parallel()
+
+	locker := &fakeAdminLocker{}
+	m := newTestLockManager(locker, time.Millisecond, 0)
+
+	release1, _ := m.Acquire("long-job")
+	m.mu.Lock()
+	m.heldSince = time.Now().Add(-time.Hour)
+	m.mu.Unlock()
+
+	acquired := make(chan func(), 1)
+	go func() {
+		release2, _ := m.Acquire("late-joiner")
+		acquired <- release2
+	}()
+
+	select {
+	case release2 := <-acquired:
+		release2()
+	case <-time.After(time.Second):
+		t.Fatalf("late acquire waited for a full release with the fairness window disabled")
+	}
+	release1()
+
+	if requests, releases := locker.counts(); requests != 1 || releases != 1 {
+		t.Fatalf("expected one shared hold, got requests=%d releases=%d", requests, releases)
+	}
+}
+
+func TestAdminLockManagerSetFairnessWindowWakesWaiters(t *testing.T) {
+	t.Parallel()
+
+	locker := &fakeAdminLocker{}
+	m := newTestLockManager(locker, time.Millisecond, time.Millisecond)
+
+	release1, _ := m.Acquire("long-job")
+	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
+	}()
+
+	require.Eventually(t, func() bool { return len(m.Status().Waiters) == 1 }, time.Second, time.Millisecond)
+	select {
+	case <-acquired:
+		t.Fatalf("late acquire joined a hold older than the fairness window")
+	default:
+	}
+
+	m.SetFairnessWindow(0)
+
+	select {
+	case release2 := <-acquired:
+		release2()
+	case <-time.After(time.Second):
+		t.Fatalf("disabling the fairness window did not admit the waiting acquire")
+	}
+	release1()
+
+	if requests, releases := locker.counts(); requests != 1 || releases != 1 {
+		t.Fatalf("expected one shared hold, got requests=%d releases=%d", requests, releases)
+	}
+}

weed/admin/dash/admin_server.go

@@ -1708,6 +1708,15 @@ func (s *AdminServer) GetPlugin() *adminplugin.Plugin {
 	return s.plugin
 }
 
+// SetAdminLockFairnessWindow sets the cluster admin lock's fairness window
+// (see AdminLockManager.SetFairnessWindow).
+func (s *AdminServer) SetAdminLockFairnessWindow(window time.Duration) {
+	if s == nil || s.pluginLock == nil {
+		return
+	}
+	s.pluginLock.SetFairnessWindow(window)
+}
+
 func (s *AdminServer) acquirePluginLock(reason string) (func(), error) {
 	if s == nil || s.pluginLock == nil {
 		return func() {}, nil

weed/command/admin.go

@@ -65,6 +65,8 @@ type AdminOptions struct {
 	cpuProfile        *string
 	memProfile        *string
 	uploadMaxSizeMB   *int
+	// nil for callers other than runAdmin; the lock keeps its default then
+	lockFairnessWindow *time.Duration
 
 	// workerGrpcListener, when set, is a listener already bound to grpcPort by
 	// the caller. `weed mini` reserves the port this way because the admin
@@ -127,6 +129,7 @@ func init() {
 	a.cpuProfile = cmdAdmin.Flag.String("cpuprofile", "", "cpu profile output file")
 	a.memProfile = cmdAdmin.Flag.String("memprofile", "", "memory profile output file")
 	a.uploadMaxSizeMB = cmdAdmin.Flag.Int("uploadMaxSizeMB", dash.DefaultUploadMaxSizeMB, "maximum total size in MB of one file browser upload (0 for no limit)")
+	a.lockFairnessWindow = cmdAdmin.Flag.Duration("adminLock.fairnessWindow", dash.DefaultAdminLockFairnessWindow, "once scheduled jobs have held the cluster admin lock this long, new jobs wait until every holder releases it so shell clients get a turn; 0 lets jobs keep joining the hold, and shell clients then get the lock when the scheduler next drains")
 }
 
 var cmdAdmin = &Command{
@@ -505,6 +508,9 @@ func startAdminServer(ctx context.Context, options AdminOptions, enableUI bool,
 	if options.uploadMaxSizeMB != nil {
 		adminServer.SetUploadMaxSize(int64(*options.uploadMaxSizeMB) << 20)
 	}
+	if options.lockFairnessWindow != nil {
+		adminServer.SetAdminLockFairnessWindow(*options.lockFairnessWindow)
+	}
 
 	if err := adminServer.ApplyPluginConfigFromToml(util.GetViper()); err != nil {
 		return fmt.Errorf("apply admin.toml to plugin config: %w", err)