fix(auth): bound force refresh all concurrency using worker pool

- Limit concurrent credential refreshes in `ForceRefreshAll` using a worker pool bounded by `AuthAutoRefreshWorkers`.
- Centralize refresh worker pool size resolution in `refreshWorkers`.
- Check context cancellation prior to refreshing to fast-fail queued credentials.

Closes: #5687

config.example.yaml

@@ -202,7 +202,7 @@ disable-image-generation: false
 # to the credential that created them. Default: 3h.
 video-result-auth-cache-ttl: "3h"
 
-# Core auth auto-refresh worker pool size (OAuth/file-based auth token refresh).
+# Core auth auto-refresh and manual refresh-all worker pool size (OAuth/file-based auth token refresh).
 # When > 0, overrides the default worker count (16).
 # auth-auto-refresh-workers: 16
 

internal/config/config.go

@@ -72,7 +72,7 @@ type Config struct {
 	// 0 keeps the legacy default cooldown. Negative values disable these cooldowns.
 	TransientErrorCooldownSeconds int `yaml:"transient-error-cooldown-seconds" json:"transient-error-cooldown-seconds"`
 
-	// AuthAutoRefreshWorkers overrides the size of the core auth auto-refresh worker pool.
+	// AuthAutoRefreshWorkers overrides the size of the core auth auto-refresh and manual refresh-all worker pool.
 	// When <= 0, the default worker count is used.
 	AuthAutoRefreshWorkers int `yaml:"auth-auto-refresh-workers" json:"auth-auto-refresh-workers"`
 

sdk/cliproxy/auth/conductor_refresh.go

@@ -55,10 +55,7 @@ func (m *Manager) StartAutoRefresh(parent context.Context, interval time.Duratio
 	}
 
 	ctx, cancelCtx := context.WithCancel(parent)
-	workers := refreshMaxConcurrency
-	if cfg, ok := m.runtimeConfig.Load().(*internalconfig.Config); ok && cfg != nil && cfg.AuthAutoRefreshWorkers > 0 {
-		workers = cfg.AuthAutoRefreshWorkers
-	}
+	workers := m.refreshWorkers()
 	loop := newAuthAutoRefreshLoop(m, interval, workers)
 
 	m.mu.Lock()
@@ -585,6 +582,16 @@ func (m *Manager) ForceRefreshAuth(ctx context.Context, id string) (*Auth, error
 	return m.refreshAuthForRequest(ctx, id, "")
 }
 
+func (m *Manager) refreshWorkers() int {
+	workers := refreshMaxConcurrency
+	if m != nil {
+		if cfg, ok := m.runtimeConfig.Load().(*internalconfig.Config); ok && cfg != nil && cfg.AuthAutoRefreshWorkers > 0 {
+			workers = cfg.AuthAutoRefreshWorkers
+		}
+	}
+	return workers
+}
+
 // ForceRefreshResult records the outcome of a forced refresh for one credential.
 type ForceRefreshResult struct {
 	ID      string `json:"id"`
@@ -597,6 +604,9 @@ func (m *Manager) ForceRefreshAll(ctx context.Context) []ForceRefreshResult {
 	if m == nil {
 		return nil
 	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	m.mu.RLock()
 	ids := make([]string, 0, len(m.auths))
 	for id, auth := range m.auths {
@@ -607,18 +617,52 @@ func (m *Manager) ForceRefreshAll(ctx context.Context) []ForceRefreshResult {
 	m.mu.RUnlock()
 
 	results := make([]ForceRefreshResult, len(ids))
-	var wg sync.WaitGroup
+	if len(ids) == 0 {
+		return results
+	}
+
+	workers := m.refreshWorkers()
+	if workers <= 0 {
+		workers = 1
+	}
+	if workers > len(ids) {
+		workers = len(ids)
+	}
+
+	type refreshJob struct {
+		index  int
+		authID string
+	}
+
+	jobCh := make(chan refreshJob, len(ids))
 	for i, id := range ids {
+		jobCh <- refreshJob{index: i, authID: id}
+	}
+	close(jobCh)
+
+	var wg sync.WaitGroup
+	for w := 0; w < workers; w++ {
 		wg.Add(1)
-		go func(index int, authID string) {
+		go func() {
 			defer wg.Done()
-			_, err := m.ForceRefreshAuth(ctx, authID)
-			res := ForceRefreshResult{ID: authID, Success: err == nil}
-			if err != nil {
-				res.Error = err.Error()
+			for job := range jobCh {
+				if errCtx := ctx.Err(); errCtx != nil {
+					results[job.index] = ForceRefreshResult{
+						ID:      job.authID,
+						Success: false,
+						Error:   errCtx.Error(),
+					}
+					continue
+				}
+
+				_, err := m.ForceRefreshAuth(ctx, job.authID)
+				res := ForceRefreshResult{ID: job.authID, Success: err == nil}
+				if err != nil {
+					res.Error = err.Error()
+				}
+				results[job.index] = res
 			}
-			results[index] = res
-		}(i, id)
+		}()
 	}
 	wg.Wait()
 	return results

sdk/cliproxy/auth/force_refresh_test.go

@@ -2,7 +2,13 @@ package auth
 
 import (
 	"context"
+	"fmt"
+	"net/http"
+	"sync/atomic"
 	"testing"
+
+	internalconfig "github.com/router-for-me/CLIProxyAPI/v7/internal/config"
+	cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
 )
 
 func TestManager_ForceRefreshAuth_ClearsErrorAndRefreshes(t *testing.T) {
@@ -127,3 +133,233 @@ func TestManager_ForceRefreshAuth_PreservesErrorOnFailure(t *testing.T) {
 		t.Fatal("LastError should not be wiped on failure")
 	}
 }
+
+type concurrencyTrackingRefreshExecutor struct {
+	id            string
+	current       atomic.Int32
+	maxConcurrent atomic.Int32
+	totalEntered  atomic.Int32
+	enteredCh     chan struct{}
+	releaseCh     chan struct{}
+}
+
+func (e *concurrencyTrackingRefreshExecutor) Identifier() string { return e.id }
+func (e *concurrencyTrackingRefreshExecutor) Execute(context.Context, *Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
+	return cliproxyexecutor.Response{}, nil
+}
+func (e *concurrencyTrackingRefreshExecutor) ExecuteStream(context.Context, *Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (*cliproxyexecutor.StreamResult, error) {
+	return nil, nil
+}
+func (e *concurrencyTrackingRefreshExecutor) CountTokens(context.Context, *Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
+	return cliproxyexecutor.Response{}, nil
+}
+func (e *concurrencyTrackingRefreshExecutor) HttpRequest(context.Context, *Auth, *http.Request) (*http.Response, error) {
+	return nil, nil
+}
+func (e *concurrencyTrackingRefreshExecutor) Refresh(_ context.Context, auth *Auth) (*Auth, error) {
+	e.totalEntered.Add(1)
+	cur := e.current.Add(1)
+	for {
+		max := e.maxConcurrent.Load()
+		if cur <= max || e.maxConcurrent.CompareAndSwap(max, cur) {
+			break
+		}
+	}
+	if e.enteredCh != nil {
+		e.enteredCh <- struct{}{}
+	}
+	if e.releaseCh != nil {
+		<-e.releaseCh
+	}
+	e.current.Add(-1)
+	if auth.Metadata == nil {
+		auth.Metadata = make(map[string]any)
+	}
+	auth.Metadata["access_token"] = "refreshed-token"
+	return auth, nil
+}
+
+func TestManager_ForceRefreshAll_WorkersBounded(t *testing.T) {
+	ctx := context.Background()
+	manager := NewManager(nil, &RoundRobinSelector{}, nil)
+	manager.runtimeConfig.Store(&internalconfig.Config{AuthAutoRefreshWorkers: 2})
+
+	executor := &concurrencyTrackingRefreshExecutor{
+		id:        "antigravity",
+		enteredCh: make(chan struct{}, 6),
+		releaseCh: make(chan struct{}),
+	}
+	manager.RegisterExecutor(executor)
+
+	for i := 0; i < 6; i++ {
+		auth := &Auth{
+			ID:       fmt.Sprintf("ag-%d", i),
+			Provider: "antigravity",
+			Metadata: map[string]any{"refresh_token": fmt.Sprintf("ref-%d", i)},
+		}
+		if _, err := manager.Register(ctx, auth); err != nil {
+			t.Fatalf("register auth: %v", err)
+		}
+	}
+
+	done := make(chan struct{})
+	var results []ForceRefreshResult
+	go func() {
+		results = manager.ForceRefreshAll(ctx)
+		close(done)
+	}()
+
+	// Wait for 2 workers to enter Refresh
+	<-executor.enteredCh
+	<-executor.enteredCh
+
+	// If unbounded, remaining goroutines also enter Refresh
+	// Release the blocked workers
+	close(executor.releaseCh)
+	<-done
+
+	if len(results) != 6 {
+		t.Fatalf("expected 6 results, got %d", len(results))
+	}
+	for _, res := range results {
+		if !res.Success {
+			t.Fatalf("expected success for %s, got error: %s", res.ID, res.Error)
+		}
+	}
+
+	if max := executor.maxConcurrent.Load(); max > 2 {
+		t.Fatalf("expected max concurrent calls <= 2, got %d", max)
+	}
+}
+
+func TestManager_ForceRefreshAll_CanceledContextSkipsExecutors(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	manager := NewManager(nil, &RoundRobinSelector{}, nil)
+	manager.runtimeConfig.Store(&internalconfig.Config{AuthAutoRefreshWorkers: 2})
+	executor := &countingRefreshExecutor{id: "antigravity"}
+	manager.RegisterExecutor(executor)
+
+	for i := 0; i < 6; i++ {
+		auth := &Auth{
+			ID:       fmt.Sprintf("ag-%d", i),
+			Provider: "antigravity",
+			Metadata: map[string]any{"refresh_token": fmt.Sprintf("ref-%d", i)},
+		}
+		if _, err := manager.Register(context.Background(), auth); err != nil {
+			t.Fatalf("register auth: %v", err)
+		}
+	}
+
+	results := manager.ForceRefreshAll(ctx)
+	if len(results) != 6 {
+		t.Fatalf("expected 6 results, got %d", len(results))
+	}
+	for _, res := range results {
+		if res.Success {
+			t.Fatalf("expected failure due to cancellation for %s", res.ID)
+		}
+		if res.Error != context.Canceled.Error() {
+			t.Fatalf("expected error %q, got %q", context.Canceled.Error(), res.Error)
+		}
+	}
+	if executor.refreshCalls.Load() != 0 {
+		t.Fatalf("expected 0 refresh calls when context canceled, got %d", executor.refreshCalls.Load())
+	}
+}
+
+func TestManager_ForceRefreshAll_DynamicCancellationSkipsRemaining(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	manager := NewManager(nil, &RoundRobinSelector{}, nil)
+	manager.runtimeConfig.Store(&internalconfig.Config{AuthAutoRefreshWorkers: 2})
+
+	executor := &concurrencyTrackingRefreshExecutor{
+		id:        "antigravity",
+		enteredCh: make(chan struct{}, 6),
+		releaseCh: make(chan struct{}),
+	}
+	manager.RegisterExecutor(executor)
+
+	for i := 0; i < 6; i++ {
+		auth := &Auth{
+			ID:       fmt.Sprintf("ag-%d", i),
+			Provider: "antigravity",
+			Metadata: map[string]any{"refresh_token": fmt.Sprintf("ref-%d", i)},
+		}
+		if _, err := manager.Register(context.Background(), auth); err != nil {
+			t.Fatalf("register auth: %v", err)
+		}
+	}
+
+	done := make(chan struct{})
+	var results []ForceRefreshResult
+	go func() {
+		results = manager.ForceRefreshAll(ctx)
+		close(done)
+	}()
+
+	// Wait for 2 workers to enter Refresh
+	<-executor.enteredCh
+	<-executor.enteredCh
+
+	// Cancel context while first batch is in flight
+	cancel()
+
+	// Release the blocked workers
+	close(executor.releaseCh)
+	<-done
+
+	if len(results) != 6 {
+		t.Fatalf("expected 6 results, got %d", len(results))
+	}
+
+	// Exactly 2 workers entered Refresh; remaining 4 were canceled in queue
+	if entered := executor.totalEntered.Load(); entered != 2 {
+		t.Fatalf("expected exactly 2 entered calls, got %d", entered)
+	}
+
+	successCount := 0
+	cancelCount := 0
+	for _, res := range results {
+		if res.Success {
+			successCount++
+		} else if res.Error == context.Canceled.Error() {
+			cancelCount++
+		}
+	}
+	if successCount != 2 {
+		t.Fatalf("expected 2 successful refreshes, got %d", successCount)
+	}
+	if cancelCount != 4 {
+		t.Fatalf("expected 4 canceled refreshes, got %d", cancelCount)
+	}
+}
+
+func TestManager_RefreshWorkersResolution(t *testing.T) {
+	manager := NewManager(nil, &RoundRobinSelector{}, nil)
+
+	// Default when config is empty
+	if w := manager.refreshWorkers(); w != 16 {
+		t.Fatalf("expected default 16 workers, got %d", w)
+	}
+
+	// Zero should keep default 16
+	manager.runtimeConfig.Store(&internalconfig.Config{AuthAutoRefreshWorkers: 0})
+	if w := manager.refreshWorkers(); w != 16 {
+		t.Fatalf("expected 16 workers for 0, got %d", w)
+	}
+
+	// Negative should keep default 16
+	manager.runtimeConfig.Store(&internalconfig.Config{AuthAutoRefreshWorkers: -5})
+	if w := manager.refreshWorkers(); w != 16 {
+		t.Fatalf("expected 16 workers for -5, got %d", w)
+	}
+
+	// Positive override
+	manager.runtimeConfig.Store(&internalconfig.Config{AuthAutoRefreshWorkers: 4})
+	if w := manager.refreshWorkers(); w != 4 {
+		t.Fatalf("expected 4 workers, got %d", w)
+	}
+}