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