From 608b7e7fa2de3e40c9677cdc0eb40909db8996aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Sat, 21 Mar 2026 09:23:58 +0800 Subject: [PATCH] fix(ccm,ocm): stop cascading 429 retry storm on token refresh When the access token expires and refreshToken() gets 429, getAccessToken() returned the error but left credentials unchanged with no cooldown. Every subsequent request re-attempted the refresh, creating a burst that overwhelmed the token endpoint. - refreshToken() now returns Retry-After duration from 429 response headers (-1 when no header present, meaning permanently blocked) - getAccessToken() caches the 429 and blocks further refresh attempts until Retry-After expires (or permanently if no header) - reloadCredentials() clears the block when new credentials are loaded from file - Remove go pollUsage() on upstream errors (unrelated to usage state) --- service/ccm/credential_default.go | 24 +++++++++++++++++++++++- service/ccm/credential_file.go | 3 +++ service/ccm/credential_oauth.go | 24 ++++++++++++++++-------- service/ccm/service_handler.go | 1 - service/ocm/credential_default.go | 24 +++++++++++++++++++++++- service/ocm/credential_file.go | 3 +++ service/ocm/credential_oauth.go | 24 ++++++++++++++++-------- service/ocm/service_handler.go | 1 - 8 files changed, 84 insertions(+), 20 deletions(-) diff --git a/service/ccm/credential_default.go b/service/ccm/credential_default.go index 1003c7b36..24549a82a 100644 --- a/service/ccm/credential_default.go +++ b/service/ccm/credential_default.go @@ -46,6 +46,11 @@ type defaultCredential struct { statusSubscriber *observable.Subscriber[struct{}] + // Refresh rate-limit cooldown (protected by access mutex) + refreshRetryAt time.Time + refreshRetryError error + refreshBlocked bool + // Connection interruption interrupted bool requestContext context.Context @@ -197,16 +202,33 @@ func (c *defaultCredential) getAccessToken() (string, error) { return c.credentials.AccessToken, nil } + if c.refreshBlocked { + return "", c.refreshRetryError + } + if !c.refreshRetryAt.IsZero() && time.Now().Before(c.refreshRetryAt) { + return "", c.refreshRetryError + } + err = platformCanWriteCredentials(c.credentialPath) if err != nil { return "", E.Cause(err, "credential file not writable, refusing refresh to avoid invalidation") } baseCredentials := cloneCredentials(c.credentials) - newCredentials, err := refreshToken(c.serviceContext, c.forwardHTTPClient, c.credentials) + newCredentials, retryDelay, err := refreshToken(c.serviceContext, c.forwardHTTPClient, c.credentials) if err != nil { + if retryDelay < 0 { + c.refreshBlocked = true + c.refreshRetryError = err + } else if retryDelay > 0 { + c.refreshRetryAt = time.Now().Add(retryDelay) + c.refreshRetryError = err + } return "", err } + c.refreshRetryAt = time.Time{} + c.refreshRetryError = nil + c.refreshBlocked = false latestCredentials, latestErr := platformReadCredentials(c.credentialPath) if latestErr == nil && !credentialsEqual(latestCredentials, baseCredentials) { diff --git a/service/ccm/credential_file.go b/service/ccm/credential_file.go index 4a6531471..3f67eaf13 100644 --- a/service/ccm/credential_file.go +++ b/service/ccm/credential_file.go @@ -108,6 +108,9 @@ func (c *defaultCredential) reloadCredentials(force bool) error { c.access.Lock() c.credentials = credentials + c.refreshRetryAt = time.Time{} + c.refreshRetryError = nil + c.refreshBlocked = false c.access.Unlock() c.stateAccess.Lock() diff --git a/service/ccm/credential_oauth.go b/service/ccm/credential_oauth.go index da559c173..114f87d3e 100644 --- a/service/ccm/credential_oauth.go +++ b/service/ccm/credential_oauth.go @@ -11,6 +11,7 @@ import ( "path/filepath" "runtime" "slices" + "strconv" "sync" "time" @@ -144,9 +145,9 @@ func (c *oauthCredentials) needsRefresh() bool { return time.Now().UnixMilli() >= c.ExpiresAt-tokenRefreshBufferMs } -func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oauthCredentials) (*oauthCredentials, error) { +func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oauthCredentials) (*oauthCredentials, time.Duration, error) { if credentials.RefreshToken == "" { - return nil, E.New("refresh token is empty") + return nil, 0, E.New("refresh token is empty") } requestBody, err := json.Marshal(map[string]string{ @@ -155,7 +156,7 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau "client_id": oauth2ClientID, }) if err != nil { - return nil, E.Cause(err, "marshal request") + return nil, 0, E.Cause(err, "marshal request") } response, err := doHTTPWithRetry(ctx, httpClient, func() (*http.Request, error) { @@ -168,17 +169,24 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau return request, nil }) if err != nil { - return nil, err + return nil, 0, err } defer response.Body.Close() if response.StatusCode == http.StatusTooManyRequests { body, _ := io.ReadAll(response.Body) - return nil, E.New("refresh rate limited: ", response.Status, " ", string(body)) + retryDelay := time.Duration(-1) + if retryAfter := response.Header.Get("Retry-After"); retryAfter != "" { + seconds, parseErr := strconv.ParseInt(retryAfter, 10, 64) + if parseErr == nil && seconds > 0 { + retryDelay = time.Duration(seconds) * time.Second + } + } + return nil, retryDelay, E.New("refresh rate limited: ", response.Status, " ", string(body)) } if response.StatusCode != http.StatusOK { body, _ := io.ReadAll(response.Body) - return nil, E.New("refresh failed: ", response.Status, " ", string(body)) + return nil, 0, E.New("refresh failed: ", response.Status, " ", string(body)) } var tokenResponse struct { @@ -188,7 +196,7 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau } err = json.NewDecoder(response.Body).Decode(&tokenResponse) if err != nil { - return nil, E.Cause(err, "decode response") + return nil, 0, E.Cause(err, "decode response") } newCredentials := *credentials @@ -198,7 +206,7 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau } newCredentials.ExpiresAt = time.Now().UnixMilli() + int64(tokenResponse.ExpiresIn)*1000 - return &newCredentials, nil + return &newCredentials, 0, nil } func cloneCredentials(credentials *oauthCredentials) *oauthCredentials { diff --git a/service/ccm/service_handler.go b/service/ccm/service_handler.go index 87f943cca..e034dc041 100644 --- a/service/ccm/service_handler.go +++ b/service/ccm/service_handler.go @@ -317,7 +317,6 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { if response.StatusCode != http.StatusOK && response.StatusCode != http.StatusTooManyRequests { body, _ := io.ReadAll(response.Body) s.logger.ErrorContext(ctx, "upstream error from ", selectedCredential.tagName(), ": status ", response.StatusCode, " ", string(body)) - go selectedCredential.pollUsage() writeJSONError(w, r, http.StatusInternalServerError, "api_error", "proxy request (status "+strconv.Itoa(response.StatusCode)+"): "+string(body)) return diff --git a/service/ocm/credential_default.go b/service/ocm/credential_default.go index 3622ac8ff..c3e8335bb 100644 --- a/service/ocm/credential_default.go +++ b/service/ocm/credential_default.go @@ -48,6 +48,11 @@ type defaultCredential struct { statusSubscriber *observable.Subscriber[struct{}] + // Refresh rate-limit cooldown (protected by access mutex) + refreshRetryAt time.Time + refreshRetryError error + refreshBlocked bool + // Connection interruption onBecameUnusable func() interrupted bool @@ -201,16 +206,33 @@ func (c *defaultCredential) getAccessToken() (string, error) { return c.credentials.getAccessToken(), nil } + if c.refreshBlocked { + return "", c.refreshRetryError + } + if !c.refreshRetryAt.IsZero() && time.Now().Before(c.refreshRetryAt) { + return "", c.refreshRetryError + } + err = platformCanWriteCredentials(c.credentialPath) if err != nil { return "", E.Cause(err, "credential file not writable, refusing refresh to avoid invalidation") } baseCredentials := cloneCredentials(c.credentials) - newCredentials, err := refreshToken(c.serviceContext, c.forwardHTTPClient, c.credentials) + newCredentials, retryDelay, err := refreshToken(c.serviceContext, c.forwardHTTPClient, c.credentials) if err != nil { + if retryDelay < 0 { + c.refreshBlocked = true + c.refreshRetryError = err + } else if retryDelay > 0 { + c.refreshRetryAt = time.Now().Add(retryDelay) + c.refreshRetryError = err + } return "", err } + c.refreshRetryAt = time.Time{} + c.refreshRetryError = nil + c.refreshBlocked = false latestCredentials, latestErr := platformReadCredentials(c.credentialPath) if latestErr == nil && !credentialsEqual(latestCredentials, baseCredentials) { diff --git a/service/ocm/credential_file.go b/service/ocm/credential_file.go index b15417a46..d5f23a7e2 100644 --- a/service/ocm/credential_file.go +++ b/service/ocm/credential_file.go @@ -108,6 +108,9 @@ func (c *defaultCredential) reloadCredentials(force bool) error { c.access.Lock() c.credentials = credentials + c.refreshRetryAt = time.Time{} + c.refreshRetryError = nil + c.refreshBlocked = false c.access.Unlock() c.stateAccess.Lock() diff --git a/service/ocm/credential_oauth.go b/service/ocm/credential_oauth.go index bb240b5ab..fd4692998 100644 --- a/service/ocm/credential_oauth.go +++ b/service/ocm/credential_oauth.go @@ -9,6 +9,7 @@ import ( "os" "os/user" "path/filepath" + "strconv" "time" E "github.com/sagernet/sing/common/exceptions" @@ -119,9 +120,9 @@ func (c *oauthCredentials) needsRefresh() bool { return time.Since(*c.LastRefresh) >= time.Duration(tokenRefreshIntervalDays)*24*time.Hour } -func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oauthCredentials) (*oauthCredentials, error) { +func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oauthCredentials) (*oauthCredentials, time.Duration, error) { if credentials.Tokens == nil || credentials.Tokens.RefreshToken == "" { - return nil, E.New("refresh token is empty") + return nil, 0, E.New("refresh token is empty") } requestBody, err := json.Marshal(map[string]string{ @@ -131,7 +132,7 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau "scope": "openid profile email", }) if err != nil { - return nil, E.Cause(err, "marshal request") + return nil, 0, E.Cause(err, "marshal request") } response, err := doHTTPWithRetry(ctx, httpClient, func() (*http.Request, error) { @@ -144,17 +145,24 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau return request, nil }) if err != nil { - return nil, err + return nil, 0, err } defer response.Body.Close() if response.StatusCode == http.StatusTooManyRequests { body, _ := io.ReadAll(response.Body) - return nil, E.New("refresh rate limited: ", response.Status, " ", string(body)) + retryDelay := time.Duration(-1) + if retryAfter := response.Header.Get("Retry-After"); retryAfter != "" { + seconds, parseErr := strconv.ParseInt(retryAfter, 10, 64) + if parseErr == nil && seconds > 0 { + retryDelay = time.Duration(seconds) * time.Second + } + } + return nil, retryDelay, E.New("refresh rate limited: ", response.Status, " ", string(body)) } if response.StatusCode != http.StatusOK { body, _ := io.ReadAll(response.Body) - return nil, E.New("refresh failed: ", response.Status, " ", string(body)) + return nil, 0, E.New("refresh failed: ", response.Status, " ", string(body)) } var tokenResponse struct { @@ -164,7 +172,7 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau } err = json.NewDecoder(response.Body).Decode(&tokenResponse) if err != nil { - return nil, E.Cause(err, "decode response") + return nil, 0, E.Cause(err, "decode response") } newCredentials := *credentials @@ -183,7 +191,7 @@ func refreshToken(ctx context.Context, httpClient *http.Client, credentials *oau now := time.Now() newCredentials.LastRefresh = &now - return &newCredentials, nil + return &newCredentials, 0, nil } func cloneCredentials(credentials *oauthCredentials) *oauthCredentials { diff --git a/service/ocm/service_handler.go b/service/ocm/service_handler.go index 8b50f748a..d4a04457a 100644 --- a/service/ocm/service_handler.go +++ b/service/ocm/service_handler.go @@ -285,7 +285,6 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { if response.StatusCode != http.StatusOK && response.StatusCode != http.StatusTooManyRequests { body, _ := io.ReadAll(response.Body) s.logger.ErrorContext(ctx, "upstream error from ", selectedCredential.tagName(), ": status ", response.StatusCode, " ", string(body)) - go selectedCredential.pollUsage() writeJSONError(w, r, http.StatusInternalServerError, "api_error", "proxy request (status "+strconv.Itoa(response.StatusCode)+"): "+string(body)) return