From 2943e8e5f022bf19e557b7a340a1d054f80dcd6f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Sat, 14 Mar 2026 14:05:10 +0800 Subject: [PATCH] ccm,ocm: mark credentials unusable on usage poll failure and trigger poll on upstream error --- service/ccm/credential_external.go | 41 ++++++++++++++++++++---------- service/ccm/credential_state.go | 37 ++++++++++++++++++--------- service/ccm/service.go | 1 + service/ocm/credential_external.go | 41 ++++++++++++++++++++---------- service/ocm/credential_state.go | 37 ++++++++++++++++++--------- service/ocm/service.go | 1 + 6 files changed, 108 insertions(+), 50 deletions(-) diff --git a/service/ccm/credential_external.go b/service/ccm/credential_external.go index 74ce6617e..d6eb4c102 100644 --- a/service/ccm/credential_external.go +++ b/service/ccm/credential_external.go @@ -239,6 +239,10 @@ func (c *externalCredential) isUsable() bool { return false } c.stateMutex.RLock() + if c.state.consecutivePollFailures > 0 { + c.stateMutex.RUnlock() + return false + } if c.state.hardRateLimited { if time.Now().Before(c.state.rateLimitResetAt) { c.stateMutex.RUnlock() @@ -402,6 +406,7 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) { } } if hadData { + c.state.consecutivePollFailures = 0 c.state.lastUpdated = time.Now() } if isFirstUpdate || int(c.state.fiveHourUtilization*100) != int(oldFiveHour*100) || int(c.state.weeklyUtilization*100) != int(oldWeekly*100) { @@ -419,7 +424,7 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) { } func (c *externalCredential) checkTransitionLocked() bool { - unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 + unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 || c.state.consecutivePollFailures > 0 if unusable && !c.interrupted { c.interrupted = true return true @@ -479,19 +484,24 @@ func (c *externalCredential) pollUsage(ctx context.Context) { }) if err != nil { c.logger.Error("poll usage for ", c.tag, ": ", err) - c.stateMutex.Lock() - c.state.consecutivePollFailures++ - c.stateMutex.Unlock() + c.incrementPollFailures() return } defer response.Body.Close() if response.StatusCode != http.StatusOK { body, _ := io.ReadAll(response.Body) - c.stateMutex.Lock() - c.state.consecutivePollFailures++ - c.stateMutex.Unlock() c.logger.Debug("poll usage for ", c.tag, ": status ", response.StatusCode, " ", string(body)) + // 404 means the remote does not have a status endpoint yet; + // usage will be updated passively from response headers. + if response.StatusCode == http.StatusNotFound { + c.stateMutex.Lock() + c.state.consecutivePollFailures = 0 + c.checkTransitionLocked() + c.stateMutex.Unlock() + } else { + c.incrementPollFailures() + } return } @@ -501,10 +511,8 @@ func (c *externalCredential) pollUsage(ctx context.Context) { } err = json.NewDecoder(response.Body).Decode(&statusResponse) if err != nil { - c.stateMutex.Lock() - c.state.consecutivePollFailures++ - c.stateMutex.Unlock() c.logger.Debug("poll usage for ", c.tag, ": decode: ", err) + c.incrementPollFailures() return } @@ -551,10 +559,17 @@ func (c *externalCredential) pollBackoff(baseInterval time.Duration) time.Durati if failures <= 0 { return baseInterval } - if failures > 4 { - failures = 4 + return failedPollRetryInterval +} + +func (c *externalCredential) incrementPollFailures() { + c.stateMutex.Lock() + c.state.consecutivePollFailures++ + shouldInterrupt := c.checkTransitionLocked() + c.stateMutex.Unlock() + if shouldInterrupt { + c.interruptConnections() } - return baseInterval * time.Duration(1< 0 { + c.stateMutex.RUnlock() + return false + } if c.state.hardRateLimited { if time.Now().Before(c.state.rateLimitResetAt) { c.stateMutex.RUnlock() @@ -459,7 +467,7 @@ func (c *defaultCredential) checkReservesLocked() bool { // checkTransitionLocked detects usable→unusable transition. // Must be called with stateMutex write lock held. func (c *defaultCredential) checkTransitionLocked() bool { - unusable := c.state.unavailable || c.state.hardRateLimited || !c.checkReservesLocked() + unusable := c.state.unavailable || c.state.hardRateLimited || !c.checkReservesLocked() || c.state.consecutivePollFailures > 0 if unusable && !c.interrupted { c.interrupted = true return true @@ -534,6 +542,16 @@ func (c *defaultCredential) markUsagePollAttempted() { c.state.lastUpdated = time.Now() } +func (c *defaultCredential) incrementPollFailures() { + c.stateMutex.Lock() + c.state.consecutivePollFailures++ + shouldInterrupt := c.checkTransitionLocked() + c.stateMutex.Unlock() + if shouldInterrupt { + c.interruptConnections() + } +} + func (c *defaultCredential) pollBackoff(baseInterval time.Duration) time.Duration { c.stateMutex.RLock() failures := c.state.consecutivePollFailures @@ -541,10 +559,7 @@ func (c *defaultCredential) pollBackoff(baseInterval time.Duration) time.Duratio if failures <= 0 { return baseInterval } - if failures > 4 { - failures = 4 - } - return baseInterval * time.Duration(1< 0 { + c.stateMutex.RUnlock() + return false + } if c.state.hardRateLimited { if time.Now().Before(c.state.rateLimitResetAt) { c.stateMutex.RUnlock() @@ -439,6 +443,7 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) { } } if hadData { + c.state.consecutivePollFailures = 0 c.state.lastUpdated = time.Now() } if isFirstUpdate || int(c.state.fiveHourUtilization*100) != int(oldFiveHour*100) || int(c.state.weeklyUtilization*100) != int(oldWeekly*100) { @@ -456,7 +461,7 @@ func (c *externalCredential) updateStateFromHeaders(headers http.Header) { } func (c *externalCredential) checkTransitionLocked() bool { - unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 + unusable := c.state.hardRateLimited || c.state.fiveHourUtilization >= 100 || c.state.weeklyUtilization >= 100 || c.state.consecutivePollFailures > 0 if unusable && !c.interrupted { c.interrupted = true return true @@ -516,19 +521,24 @@ func (c *externalCredential) pollUsage(ctx context.Context) { }) if err != nil { c.logger.Error("poll usage for ", c.tag, ": ", err) - c.stateMutex.Lock() - c.state.consecutivePollFailures++ - c.stateMutex.Unlock() + c.incrementPollFailures() return } defer response.Body.Close() if response.StatusCode != http.StatusOK { body, _ := io.ReadAll(response.Body) - c.stateMutex.Lock() - c.state.consecutivePollFailures++ - c.stateMutex.Unlock() c.logger.Debug("poll usage for ", c.tag, ": status ", response.StatusCode, " ", string(body)) + // 404 means the remote does not have a status endpoint yet; + // usage will be updated passively from response headers. + if response.StatusCode == http.StatusNotFound { + c.stateMutex.Lock() + c.state.consecutivePollFailures = 0 + c.checkTransitionLocked() + c.stateMutex.Unlock() + } else { + c.incrementPollFailures() + } return } @@ -538,10 +548,8 @@ func (c *externalCredential) pollUsage(ctx context.Context) { } err = json.NewDecoder(response.Body).Decode(&statusResponse) if err != nil { - c.stateMutex.Lock() - c.state.consecutivePollFailures++ - c.stateMutex.Unlock() c.logger.Debug("poll usage for ", c.tag, ": decode: ", err) + c.incrementPollFailures() return } @@ -588,10 +596,17 @@ func (c *externalCredential) pollBackoff(baseInterval time.Duration) time.Durati if failures <= 0 { return baseInterval } - if failures > 4 { - failures = 4 + return failedPollRetryInterval +} + +func (c *externalCredential) incrementPollFailures() { + c.stateMutex.Lock() + c.state.consecutivePollFailures++ + shouldInterrupt := c.checkTransitionLocked() + c.stateMutex.Unlock() + if shouldInterrupt { + c.interruptConnections() } - return baseInterval * time.Duration(1< 0 { + c.stateMutex.RUnlock() + return false + } if c.state.hardRateLimited { if time.Now().Before(c.state.rateLimitResetAt) { c.stateMutex.RUnlock() @@ -476,7 +484,7 @@ func (c *defaultCredential) checkReservesLocked() bool { // checkTransitionLocked detects usable→unusable transition. // Must be called with stateMutex write lock held. func (c *defaultCredential) checkTransitionLocked() bool { - unusable := c.state.unavailable || c.state.hardRateLimited || !c.checkReservesLocked() + unusable := c.state.unavailable || c.state.hardRateLimited || !c.checkReservesLocked() || c.state.consecutivePollFailures > 0 if unusable && !c.interrupted { c.interrupted = true return true @@ -551,6 +559,16 @@ func (c *defaultCredential) markUsagePollAttempted() { c.state.lastUpdated = time.Now() } +func (c *defaultCredential) incrementPollFailures() { + c.stateMutex.Lock() + c.state.consecutivePollFailures++ + shouldInterrupt := c.checkTransitionLocked() + c.stateMutex.Unlock() + if shouldInterrupt { + c.interruptConnections() + } +} + func (c *defaultCredential) pollBackoff(baseInterval time.Duration) time.Duration { c.stateMutex.RLock() failures := c.state.consecutivePollFailures @@ -558,10 +576,7 @@ func (c *defaultCredential) pollBackoff(baseInterval time.Duration) time.Duratio if failures <= 0 { return baseInterval } - if failures > 4 { - failures = 4 - } - return baseInterval * time.Duration(1<