wangjia/sing-box
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ccm,ocm): handle upstream 400 by marking external credentials rejected and polling default credentials

Browse Source 
External credentials returning 400 are marked unavailable for pollInterval
duration; status stream/poll success clears the rejection early. Default
credentials trigger a stale poll to let the usage API detect account issues
without causing 429 storms.
This commit is contained in:
世界
2026-03-21 10:31:17 +08:00
parent 608b7e7fa2
commit 99d9e06dd0
11 changed files with 104 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/ccm/credential.go
+2
View File
@@ -66,6 +66,7 @@ type credentialState struct {
consecutivePollFailures int
usageAPIRetryDelay time.Duration
unavailable bool
upstreamRejectedUntil time.Time
lastCredentialLoadAttempt time.Time
lastCredentialLoadError string
}
@@ -108,6 +109,7 @@ type Credential interface {
fiveHourResetTime() time.Time
weeklyResetTime() time.Time
markRateLimited(resetAt time.Time)
markUpstreamRejected()
earliestReset() time.Time
unavailableError() error
service/ccm/credential_default.go
+2
View File
@@ -354,6 +354,8 @@ func (c *defaultCredential) markRateLimited(resetAt time.Time) {
c.emitStatusUpdate()
}
func (c *defaultCredential) markUpstreamRejected() {}
func (c *defaultCredential) isUsable() bool {
c.retryCredentialReloadIfNeeded()
service/ccm/credential_external.go
+21 -1
View File
@@ -273,6 +273,10 @@ func (c *externalCredential) isUsable() bool {
c.stateAccess.RUnlock()
return false
}
if !c.state.upstreamRejectedUntil.IsZero() && time.Now().Before(c.state.upstreamRejectedUntil) {
c.stateAccess.RUnlock()
return false
}
if c.state.hardRateLimited {
if time.Now().Before(c.state.rateLimitResetAt) {
c.stateAccess.RUnlock()
@@ -347,6 +351,18 @@ func (c *externalCredential) markRateLimited(resetAt time.Time) {
c.emitStatusUpdate()
}
func (c *externalCredential) markUpstreamRejected() {
c.logger.Warn("upstream rejected credential ", c.tag, ", marking unavailable for ", log.FormatDuration(c.pollInterval))
c.stateAccess.Lock()
c.state.upstreamRejectedUntil = time.Now().Add(c.pollInterval)
shouldInterrupt := c.checkTransitionLocked()
c.stateAccess.Unlock()
if shouldInterrupt {
c.interruptConnections()
}
c.emitStatusUpdate()
}
func (c *externalCredential) earliestReset() time.Time {
c.stateAccess.RLock()
defer c.stateAccess.RUnlock()
@@ -475,6 +491,7 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) {
}
if hadData {
c.state.consecutivePollFailures = 0
c.state.upstreamRejectedUntil = time.Time{}
c.state.lastUpdated = time.Now()
}
if isFirstUpdate || int(c.state.fiveHourUtilization*100) != int(oldFiveHour*100) || int(c.state.weeklyUtilization*100) != int(oldWeekly*100) {
@@ -499,7 +516,8 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) {
}
func (c *externalCredential) checkTransitionLocked() bool {
unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 || c.state.consecutivePollFailures > 0
upstreamRejected := !c.state.upstreamRejectedUntil.IsZero() && time.Now().Before(c.state.upstreamRejectedUntil)
unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 || c.state.consecutivePollFailures > 0 || upstreamRejected
if unusable && !c.interrupted {
c.interrupted = true
return true
@@ -636,6 +654,7 @@ func (c *externalCredential) pollUsage() {
oldFiveHour := c.state.fiveHourUtilization
oldWeekly := c.state.weeklyUtilization
c.state.consecutivePollFailures = 0
c.state.upstreamRejectedUntil = time.Time{}
c.state.fiveHourUtilization = statusResponse.FiveHourUtilization
c.state.weeklyUtilization = statusResponse.WeeklyUtilization
if statusResponse.PlanWeight > 0 {
@@ -744,6 +763,7 @@ func (c *externalCredential) connectStatusStream(ctx context.Context) (statusStr
oldFiveHour := c.state.fiveHourUtilization
oldWeekly := c.state.weeklyUtilization
c.state.consecutivePollFailures = 0
c.state.upstreamRejectedUntil = time.Time{}
c.state.fiveHourUtilization = statusResponse.FiveHourUtilization
c.state.weeklyUtilization = statusResponse.WeeklyUtilization
if statusResponse.PlanWeight > 0 {
service/ccm/credential_provider.go
+13
View File
@@ -17,6 +17,7 @@ type credentialProvider interface {
onRateLimited(sessionID string, credential Credential, resetAt time.Time, selection credentialSelection) Credential
linkProviderInterrupt(credential Credential, selection credentialSelection, onInterrupt func()) func() bool
pollIfStale()
pollCredentialIfStale(credential Credential)
allCredentials() []Credential
close()
}
@@ -83,6 +84,12 @@ func (p *singleCredentialProvider) linkProviderInterrupt(_ Credential, _ credent
}
}
func (p *singleCredentialProvider) pollCredentialIfStale(credential Credential) {
if time.Since(credential.lastUpdatedTime()) > credential.pollBackoff(defaultPollInterval) {
credential.pollUsage()
}
}
func (p *singleCredentialProvider) close() {}
type sessionEntry struct {
@@ -382,6 +389,12 @@ func (p *balancerProvider) pollIfStale() {
}
}
func (p *balancerProvider) pollCredentialIfStale(credential Credential) {
if time.Since(credential.lastUpdatedTime()) > credential.pollBackoff(p.pollInterval) {
credential.pollUsage()
}
}
func (p *balancerProvider) allCredentials() []Credential {
return p.credentials
}
service/ccm/service_handler.go
+11
View File
@@ -314,6 +314,17 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
if response.StatusCode == http.StatusBadRequest {
if selectedCredential.isExternal() {
selectedCredential.markUpstreamRejected()
} else {
provider.pollCredentialIfStale(selectedCredential)
}
s.logger.ErrorContext(ctx, "upstream rejected from ", selectedCredential.tagName(), ": status ", response.StatusCode)
writeCredentialUnavailableError(w, r, provider, selectedCredential, selection, "upstream rejected credential")
return
}
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))
service/ocm/credential.go
+2
View File
@@ -67,6 +67,7 @@ type credentialState struct {
consecutivePollFailures int
usageAPIRetryDelay time.Duration
unavailable bool
upstreamRejectedUntil time.Time
lastCredentialLoadAttempt time.Time
lastCredentialLoadError string
}
@@ -109,6 +110,7 @@ type Credential interface {
weeklyResetTime() time.Time
fiveHourResetTime() time.Time
markRateLimited(resetAt time.Time)
markUpstreamRejected()
earliestReset() time.Time
unavailableError() error
service/ocm/credential_default.go
+2
View File
@@ -394,6 +394,8 @@ func (c *defaultCredential) markRateLimited(resetAt time.Time) {
c.emitStatusUpdate()
}
func (c *defaultCredential) markUpstreamRejected() {}
func (c *defaultCredential) isUsable() bool {
c.retryCredentialReloadIfNeeded()
service/ocm/credential_external.go
+21 -1
View File
@@ -298,6 +298,10 @@ func (c *externalCredential) isUsable() bool {
c.stateAccess.RUnlock()
return false
}
if !c.state.upstreamRejectedUntil.IsZero() && time.Now().Before(c.state.upstreamRejectedUntil) {
c.stateAccess.RUnlock()
return false
}
if c.state.hardRateLimited {
if time.Now().Before(c.state.rateLimitResetAt) {
c.stateAccess.RUnlock()
@@ -371,6 +375,18 @@ func (c *externalCredential) markRateLimited(resetAt time.Time) {
c.emitStatusUpdate()
}
func (c *externalCredential) markUpstreamRejected() {
c.logger.Warn("upstream rejected credential ", c.tag, ", marking unavailable for ", log.FormatDuration(c.pollInterval))
c.stateAccess.Lock()
c.state.upstreamRejectedUntil = time.Now().Add(c.pollInterval)
shouldInterrupt := c.checkTransitionLocked()
c.stateAccess.Unlock()
if shouldInterrupt {
c.interruptConnections()
}
c.emitStatusUpdate()
}
func (c *externalCredential) earliestReset() time.Time {
c.stateAccess.RLock()
defer c.stateAccess.RUnlock()
@@ -514,6 +530,7 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) {
}
if hadData {
c.state.consecutivePollFailures = 0
c.state.upstreamRejectedUntil = time.Time{}
c.state.lastUpdated = time.Now()
}
if isFirstUpdate || int(c.state.fiveHourUtilization*100) != int(oldFiveHour*100) || int(c.state.weeklyUtilization*100) != int(oldWeekly*100) {
@@ -538,7 +555,8 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) {
}
func (c *externalCredential) checkTransitionLocked() bool {
unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 || c.state.consecutivePollFailures > 0
upstreamRejected := !c.state.upstreamRejectedUntil.IsZero() && time.Now().Before(c.state.upstreamRejectedUntil)
unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 || c.state.consecutivePollFailures > 0 || upstreamRejected
if unusable && !c.interrupted {
c.interrupted = true
return true
@@ -678,6 +696,7 @@ func (c *externalCredential) pollUsage() {
oldFiveHour := c.state.fiveHourUtilization
oldWeekly := c.state.weeklyUtilization
c.state.consecutivePollFailures = 0
c.state.upstreamRejectedUntil = time.Time{}
c.state.fiveHourUtilization = statusResponse.FiveHourUtilization
c.state.weeklyUtilization = statusResponse.WeeklyUtilization
if statusResponse.FiveHourReset > 0 {
@@ -786,6 +805,7 @@ func (c *externalCredential) connectStatusStream(ctx context.Context) (statusStr
oldFiveHour := c.state.fiveHourUtilization
oldWeekly := c.state.weeklyUtilization
c.state.consecutivePollFailures = 0
c.state.upstreamRejectedUntil = time.Time{}
c.state.fiveHourUtilization = statusResponse.FiveHourUtilization
c.state.weeklyUtilization = statusResponse.WeeklyUtilization
if statusResponse.FiveHourReset > 0 {
service/ocm/credential_provider.go
+13
View File
@@ -17,6 +17,7 @@ type credentialProvider interface {
onRateLimited(sessionID string, credential Credential, resetAt time.Time, selection credentialSelection) Credential
linkProviderInterrupt(credential Credential, selection credentialSelection, onInterrupt func()) func() bool
pollIfStale()
pollCredentialIfStale(credential Credential)
allCredentials() []Credential
close()
}
@@ -83,6 +84,12 @@ func (p *singleCredentialProvider) linkProviderInterrupt(_ Credential, _ credent
}
}
func (p *singleCredentialProvider) pollCredentialIfStale(credential Credential) {
if time.Since(credential.lastUpdatedTime()) > credential.pollBackoff(defaultPollInterval) {
credential.pollUsage()
}
}
func (p *singleCredentialProvider) close() {}
type sessionEntry struct {
@@ -409,6 +416,12 @@ func (p *balancerProvider) pollIfStale() {
}
}
func (p *balancerProvider) pollCredentialIfStale(credential Credential) {
if time.Since(credential.lastUpdatedTime()) > credential.pollBackoff(p.pollInterval) {
credential.pollUsage()
}
}
func (p *balancerProvider) allCredentials() []Credential {
return p.credentials
}
service/ocm/service_handler.go
+11
View File
@@ -282,6 +282,17 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
selectedCredential.updateStateFromHeaders(response.Header)
if response.StatusCode == http.StatusBadRequest {
if selectedCredential.isExternal() {
selectedCredential.markUpstreamRejected()
} else {
provider.pollCredentialIfStale(selectedCredential)
}
s.logger.ErrorContext(ctx, "upstream rejected from ", selectedCredential.tagName(), ": status ", response.StatusCode)
writeCredentialUnavailableError(w, r, provider, selectedCredential, selection, "upstream rejected credential")
return
}
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))
service/ocm/service_websocket.go
+6
View File
@@ -264,6 +264,12 @@ func (s *Service) handleWebSocket(
selectedCredential = nextCredential
continue
}
if statusCode == http.StatusBadRequest && selectedCredential.isExternal() {
selectedCredential.markUpstreamRejected()
s.logger.ErrorContext(ctx, "upstream rejected websocket from ", selectedCredential.tagName(), ": status ", statusCode)
writeCredentialUnavailableError(w, r, provider, selectedCredential, selection, "upstream rejected credential")
return
}
if statusCode > 0 && statusResponseBody != "" {
s.logger.ErrorContext(ctx, "dial upstream websocket: status ", statusCode, " body: ", statusResponseBody)
} else {