优化

internal/service/cloud115/client.go

@@ -150,6 +150,7 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map
 		if err != nil {
 			return nil, err
 		}
+		attemptedAccess := strings.TrimPrefix(req.Header.Get("Authorization"), "Bearer ")
 
 		resp, err := c.HTTP.Do(req)
 		if err != nil {
@@ -218,7 +219,7 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map
 		// refresh_token 刷新后重试一次。刷新失败或重试后仍失败才返回，
 		// 避免长时间同步因 token 过期而整体失败。
 		if isTokenCode(base.Code) {
-			if access && c.tryRefreshTokenLocked(ctx) {
+			if access && c.tryRefreshTokenLocked(ctx, attemptedAccess) {
 				continue
 			}
 			if access {
@@ -305,21 +306,23 @@ func (c *OpenClient) doAuthJSONWithUA(ctx context.Context, method, rawURL string
 // tryRefreshTokenLocked 并发安全地刷新 access_token；成功返回 true（调用方
 // 应使用内存中的新 token 重试原请求）。
 //
-// 拿到写锁后在锁内读取 oldAccess，与持锁期间的当前值对比：若已被其他
-// goroutine 刷新过则直接复用新 token，避免并发请求连环轮转消耗 115 的
-// 一次性 refresh_token。全程持写锁读写 token 字段，无 TOCTOU 窗口。
+// failedAccess 是失败请求实际携带的 token。拿到写锁后与当前 token 对比：
+// 若已被其他 goroutine 刷新过则直接复用，避免并发请求连环轮转消耗 115
+// 的一次性 refresh_token。全程持写锁读写 token 字段，无 TOCTOU 窗口。
 //
 // 对"refresh_token 本身已失效/被吊销"（IsRefreshTokenDead，如 40140114/116/119/120）
 // 这类不可恢复的错误直接放弃并清空内存 token（提示需重新授权）。
 // 对其它失败（网络瞬时抖动、刷新接口可重试错误码等）做指数退避重试几次再放弃，
 // 避免同步长任务中途 token 到期时恰好撞上一个短暂的刷新失败就整体失败。
-func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context) bool {
+func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context, failedAccess string) bool {
 	c.tokenMu.Lock()
-	// 在已持有写锁内读取当前 token 作为"刷新前快照"，消除双重加锁窗口：
-	// 若在拿锁期间已有其他 goroutine 完成刷新，refreshTokenWhileLocked
-	// 内的 c.AccessToken != oldAccess 判断会立即命中并返回复用。
-	oldAccess := c.AccessToken
-	newToken, ok := c.refreshTokenWhileLocked(ctx, oldAccess)
+	// 请求发出后若其他 goroutine 已经刷新完成，直接复用新 token 重试；
+	// 不能再次轮换一次性的 refresh_token。
+	if failedAccess != "" && c.AccessToken != failedAccess {
+		c.tokenMu.Unlock()
+		return true
+	}
+	newToken, ok := c.refreshTokenWhileLocked(ctx, failedAccess)
 	c.tokenMu.Unlock()
 	// 回调必须在 tokenMu 释放后调用，避免上层在回调内访问客户端时死锁
 	if ok && newToken != nil && c.OnTokenRefreshed != nil {

internal/service/cloud115/cloud115_test.go

@@ -140,6 +140,47 @@ func TestRefreshTokenDead(t *testing.T) {
 	}
 }
 
+func TestConcurrentTokenFailuresShareOneRefresh(t *testing.T) {
+	var mu sync.Mutex
+	refreshCalls := 0
+	mockAPI(t, func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/open/refreshToken" {
+			t.Errorf("unexpected path %s", r.URL.Path)
+		}
+		mu.Lock()
+		refreshCalls++
+		mu.Unlock()
+		time.Sleep(50 * time.Millisecond)
+		_, _ = w.Write([]byte(`{"state":true,"data":{"access_token":"at-new","refresh_token":"rt-new","expires_in":7200}}`))
+	})
+
+	client := NewOpenClient("100195129", "at-old", "rt-old")
+	var wg sync.WaitGroup
+	results := make(chan bool, 2)
+	for range 2 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			results <- client.tryRefreshTokenLocked(context.Background(), "at-old")
+		}()
+	}
+	wg.Wait()
+	close(results)
+	for ok := range results {
+		if !ok {
+			t.Fatal("concurrent refresh should reuse the refreshed token")
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if refreshCalls != 1 {
+		t.Fatalf("refresh calls = %d, want 1", refreshCalls)
+	}
+	if client.CurrentAccessToken() != "at-new" {
+		t.Fatalf("access token = %q, want at-new", client.CurrentAccessToken())
+	}
+}
+
 func TestFsListAndDownload(t *testing.T) {
 	mockAPI(t, func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {

internal/service/cloud115/open.go

@@ -477,24 +477,32 @@ func (c *OpenClient) GetToken(qrCode *QrCodeDataReturn) (*TokenData, error) {
 
 // RefreshToken 刷新访问令牌。
 func (c *OpenClient) RefreshToken(refreshToken string) (*TokenData, error) {
+	c.tokenMu.Lock()
 	if refreshToken == "" {
-		refreshToken = c.currentRefreshToken()
+		refreshToken = c.RefreshTokenStr
 	}
 	if refreshToken == "" {
+		c.tokenMu.Unlock()
 		return nil, fmt.Errorf("没有可用的 refresh_token")
 	}
 	token, err := c.doRefreshToken(refreshToken)
 	if err != nil {
 		// refresh_token 已失效时清空内存令牌（提示需重新授权）
 		if IsRefreshTokenDead(err) {
-			c.SetAuthToken("", "")
+			c.setAuthTokenLocked("", "")
 		}
+		c.tokenMu.Unlock()
 		return nil, err
 	}
 	if token.AccessToken == "" || token.RefreshToken == "" {
+		c.tokenMu.Unlock()
 		return nil, fmt.Errorf("115: 刷新返回空凭证（access_token/refresh_token 缺失）")
 	}
-	c.SetAuthToken(token.AccessToken, token.RefreshToken)
+	c.setAuthTokenLocked(token.AccessToken, token.RefreshToken)
+	c.tokenMu.Unlock()
+	if c.OnTokenRefreshed != nil {
+		c.OnTokenRefreshed(token.AccessToken, token.RefreshToken)
+	}
 	return token, nil
 }
 

internal/service/strm_115_oauth.go

@@ -293,7 +293,11 @@ func (s *StrmService) save115OAuthToken(ctx context.Context, session *strm115Aut
 	acct.LastTestAt = &now
 	acct.LastTestResult = "授权成功"
 	acct.LastTestOK = true
-	return s.repo.StrmAccount.Update(ctx, acct)
+	if err := s.repo.StrmAccount.Update(ctx, acct); err != nil {
+		return err
+	}
+	s.invalidate115Provider(acct.ID)
+	return nil
 }
 
 func (s *StrmService) drop115AuthSession(sessionID string) {
@@ -382,8 +386,15 @@ func (s *StrmService) refresh115TokensOnce(ctx context.Context) {
 				}
 			}
 		}
-		client := cloud115.NewOpenClient(cfg["app_id"], cfg["access_token"], cfg["refresh_token"])
-		token, err := client.RefreshToken(cfg["refresh_token"])
+		provider, err := s.providerFor(ctx, acct)
+		if err != nil {
+			continue
+		}
+		openProvider, ok := provider.(interface{ OpenClient() *cloud115.OpenClient })
+		if !ok || openProvider.OpenClient() == nil {
+			continue
+		}
+		token, err := openProvider.OpenClient().RefreshToken("")
 		if err != nil {
 			msg := "令牌刷新失败：" + err.Error()
 			if cloud115.IsRefreshTokenDead(err) {

internal/service/strm_account_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/truewhile/MeBox/internal/config"
 	"github.com/truewhile/MeBox/internal/model"
 	"github.com/truewhile/MeBox/internal/repository"
+	"github.com/truewhile/MeBox/internal/service/cloud115"
 )
 
 func TestStrmAccountConfigPreviewOf(t *testing.T) {
@@ -71,6 +72,90 @@ func TestUpdateStrmAccountMergesConfigWithoutClearingSecrets(t *testing.T) {
 	}
 }
 
+func TestProviderForReuses115ClientUntilCredentialsChange(t *testing.T) {
+	svc := testStrmService(t)
+	ctx := context.Background()
+	acct, err := svc.CreateStrmAccount(ctx, "115", model.StrmProvider115, map[string]string{
+		"app_id":        "100195129",
+		"access_token":  "at-1",
+		"refresh_token": "rt-1",
+	})
+	if err != nil {
+		t.Fatalf("create account: %v", err)
+	}
+
+	first, err := svc.providerFor(ctx, acct)
+	if err != nil {
+		t.Fatalf("first provider: %v", err)
+	}
+	second, err := svc.providerFor(ctx, acct)
+	if err != nil {
+		t.Fatalf("second provider: %v", err)
+	}
+	if first != second {
+		t.Fatal("115 provider should be shared per account")
+	}
+
+	updated, err := svc.UpdateStrmAccount(ctx, acct.ID, "", nil, map[string]string{
+		"access_token":  "at-2",
+		"refresh_token": "rt-2",
+	})
+	if err != nil {
+		t.Fatalf("update account: %v", err)
+	}
+	third, err := svc.providerFor(ctx, updated)
+	if err != nil {
+		t.Fatalf("provider after credential update: %v", err)
+	}
+	if first == third {
+		t.Fatal("credential update must invalidate the shared provider")
+	}
+
+	oldClient := first.(interface{ OpenClient() *cloud115.OpenClient }).OpenClient()
+	oldClient.OnTokenRefreshed("at-stale", "rt-stale")
+	fresh, err := svc.repo.StrmAccount.FindByID(ctx, acct.ID)
+	if err != nil || fresh == nil {
+		t.Fatalf("reload account: %v", err)
+	}
+	freshCfg, err := svc.strmAccountConfig(fresh)
+	if err != nil {
+		t.Fatalf("decode refreshed account: %v", err)
+	}
+	if freshCfg["access_token"] != "at-2" || freshCfg["refresh_token"] != "rt-2" {
+		t.Fatalf("stale client overwrote new credentials: %#v", freshCfg)
+	}
+}
+
+func TestPersist115TokensKeepsSharedClient(t *testing.T) {
+	svc := testStrmService(t)
+	ctx := context.Background()
+	acct, err := svc.CreateStrmAccount(ctx, "115", model.StrmProvider115, map[string]string{
+		"app_id":        "100195129",
+		"access_token":  "at-1",
+		"refresh_token": "rt-1",
+	})
+	if err != nil {
+		t.Fatalf("create account: %v", err)
+	}
+	first, err := svc.providerFor(ctx, acct)
+	if err != nil {
+		t.Fatalf("first provider: %v", err)
+	}
+
+	svc.persist115Tokens(acct.ID, "at-2", "rt-2")
+	fresh, err := svc.repo.StrmAccount.FindByID(ctx, acct.ID)
+	if err != nil || fresh == nil {
+		t.Fatalf("reload account: %v", err)
+	}
+	second, err := svc.providerFor(ctx, fresh)
+	if err != nil {
+		t.Fatalf("second provider: %v", err)
+	}
+	if first != second {
+		t.Fatal("automatic token persistence must keep the in-memory shared client")
+	}
+}
+
 func TestDeleteStrmAccountCascadesEmbyMounts(t *testing.T) {
 	ctx := context.Background()
 	db := newServiceTestDB(t, &model.StrmAccount{}, &model.EmbyMount{}, &model.StrmSyncPath{})

internal/service/strm_service.go

@@ -95,6 +95,9 @@ type StrmService struct {
 	oauthSessions map[string]*strm115AuthSession
 	wafUntil      time.Time // 115 风控/限流熔断截止时间（由 mu 保护）
 
+	providerMu       sync.Mutex
+	provider115Cache map[string]cloud.Provider // account ID -> shared provider/OpenClient
+
 	downloadSem115  chan struct{} // 115 换直链+下载并发上限（风控兜底）
 	downloadSemDAV  chan struct{} // WebDAV/OpenList/CloudDrive2 元数据下载并发上限
 	downloadSemOnce sync.Once
@@ -158,15 +161,16 @@ func (s *StrmService) releaseDownloadSlot(provider string) {
 // NewStrmService constructs the STRM service.
 func NewStrmService(cfg *config.Config, log *zap.Logger, repos *repository.Container, crypto *CryptoService) *StrmService {
 	return &StrmService{
-		log:           log,
-		repo:          repos,
-		cfg:           cfg,
-		crypto:        crypto,
-		http:          &http.Client{Timeout: 90 * time.Second},
-		stopCh:        make(chan struct{}),
-		baseCtx:       context.Background(),
-		running:       map[string]context.CancelFunc{},
-		oauthSessions: map[string]*strm115AuthSession{},
+		log:              log,
+		repo:             repos,
+		cfg:              cfg,
+		crypto:           crypto,
+		http:             &http.Client{Timeout: 90 * time.Second},
+		stopCh:           make(chan struct{}),
+		baseCtx:          context.Background(),
+		running:          map[string]context.CancelFunc{},
+		oauthSessions:    map[string]*strm115AuthSession{},
+		provider115Cache: map[string]cloud.Provider{},
 	}
 }
 
@@ -409,6 +413,9 @@ func (s *StrmService) UpdateStrmAccount(ctx context.Context, id, name string, en
 	if err := s.repo.StrmAccount.Update(ctx, acct); err != nil {
 		return nil, err
 	}
+	if acct.Provider == model.StrmProvider115 && len(config) > 0 {
+		s.invalidate115Provider(acct.ID)
+	}
 	return acct, nil
 }
 
@@ -426,6 +433,7 @@ func (s *StrmService) DeleteStrmAccount(ctx context.Context, id string) error {
 	if err := s.repo.StrmAccount.Delete(ctx, id); err != nil {
 		return err
 	}
+	s.invalidate115Provider(id)
 	// 级联清理远程 Emby 挂载：否则留下孤儿挂载，挂载计数/列表仍会显示。
 	// 账号已删，挂载清理失败只记日志，不让删除请求报错。
 	if _, err := s.repo.EmbyMount.DeleteByAccountID(ctx, id); err != nil && s.log != nil {
@@ -464,6 +472,23 @@ func (s *StrmService) ListAccounts(ctx context.Context) ([]model.StrmAccount, er
 
 // providerFor 依据账号配置构建网盘驱动。
 func (s *StrmService) providerFor(ctx context.Context, acct *model.StrmAccount) (cloud.Provider, error) {
+	if acct != nil && acct.Provider == model.StrmProvider115 {
+		s.providerMu.Lock()
+		defer s.providerMu.Unlock()
+		if provider := s.provider115Cache[acct.ID]; provider != nil {
+			return provider, nil
+		}
+		provider, err := s.newProvider(ctx, acct)
+		if err != nil {
+			return nil, err
+		}
+		s.provider115Cache[acct.ID] = provider
+		return provider, nil
+	}
+	return s.newProvider(ctx, acct)
+}
+
+func (s *StrmService) newProvider(ctx context.Context, acct *model.StrmAccount) (cloud.Provider, error) {
 	cfg, err := s.strmAccountConfig(acct)
 	if err != nil {
 		return nil, err
@@ -482,13 +507,33 @@ func (s *StrmService) providerFor(ctx context.Context, acct *model.StrmAccount)
 	// refresh_token 再刷（一次性轮转），两者互相作废，最终把有效账号
 	// 标成“授权已失效”。
 	if oc, ok := provider.(interface{ OpenClient() *cloud115.OpenClient }); ok {
-		oc.OpenClient().OnTokenRefreshed = func(accessToken, refreshToken string) {
+		client := oc.OpenClient()
+		client.OnTokenRefreshed = func(accessToken, refreshToken string) {
+			// 账号重新授权/修改凭据后，旧客户端可能仍有在途请求。旧请求
+			// 刷新的令牌不能覆盖新授权写入的凭据。
+			if !s.isCurrent115Client(acct.ID, client) {
+				return
+			}
 			s.persist115Tokens(acct.ID, accessToken, refreshToken)
 		}
 	}
 	return provider, nil
 }
 
+func (s *StrmService) invalidate115Provider(accountID string) {
+	s.providerMu.Lock()
+	delete(s.provider115Cache, accountID)
+	s.providerMu.Unlock()
+}
+
+func (s *StrmService) isCurrent115Client(accountID string, client *cloud115.OpenClient) bool {
+	s.providerMu.Lock()
+	defer s.providerMu.Unlock()
+	provider := s.provider115Cache[accountID]
+	openProvider, ok := provider.(interface{ OpenClient() *cloud115.OpenClient })
+	return ok && openProvider.OpenClient() == client
+}
+
 // ─── 全局设置 ──────────────────────────────────────────────────────────────────
 
 // GetStrmSettings 返回全局 strm 设置（含默认值）。