172 lines
3.5 KiB
Go
172 lines
3.5 KiB
Go
package upstream
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Authentication failures intentionally use the slow retry class while
|
|
// transient network failures use capped backoff; manual disconnect closes stopCh.
|
|
|
|
func (p *connectionPool) startSupervisor() {
|
|
p.supervisorOnce.Do(func() {
|
|
go p.superviseReconnects()
|
|
})
|
|
}
|
|
|
|
func (p *connectionPool) stopped() bool {
|
|
select {
|
|
case <-p.stopCh:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (p *connectionPool) signalReconnect() {
|
|
select {
|
|
case <-p.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
select {
|
|
case p.reconnectSignal <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (p *connectionPool) scheduleReconnect(stateErr error) {
|
|
select {
|
|
case <-p.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
p.mu.Lock()
|
|
now := time.Now().UTC()
|
|
if !p.nextReconnectAt.IsZero() && p.nextReconnectAt.After(now) {
|
|
p.mu.Unlock()
|
|
return
|
|
}
|
|
p.reconnectCount++
|
|
p.lastReconnectAttemptAt = now
|
|
p.lastErrorCategory = connectionErrorCategory(stateErr)
|
|
delay := reconnectDelay(p.reconnectCount, p.lastErrorCategory)
|
|
p.nextReconnectAt = p.lastReconnectAttemptAt.Add(delay)
|
|
p.mu.Unlock()
|
|
p.signalReconnect()
|
|
}
|
|
|
|
func (p *connectionPool) resetReconnectState() {
|
|
p.mu.Lock()
|
|
p.reconnectCount = 0
|
|
p.lastReconnectAttemptAt = time.Time{}
|
|
p.nextReconnectAt = time.Time{}
|
|
p.lastErrorCategory = ""
|
|
p.mu.Unlock()
|
|
p.signalReconnect()
|
|
}
|
|
|
|
func (p *connectionPool) superviseReconnects() {
|
|
for {
|
|
select {
|
|
case <-p.stopCh:
|
|
return
|
|
case <-p.reconnectSignal:
|
|
}
|
|
for {
|
|
p.mu.Lock()
|
|
next := p.nextReconnectAt
|
|
p.mu.Unlock()
|
|
if next.IsZero() {
|
|
break
|
|
}
|
|
timer := time.NewTimer(time.Until(next))
|
|
select {
|
|
case <-p.stopCh:
|
|
if !timer.Stop() {
|
|
<-timer.C
|
|
}
|
|
return
|
|
case <-p.reconnectSignal:
|
|
if !timer.Stop() {
|
|
<-timer.C
|
|
}
|
|
continue
|
|
case <-timer.C:
|
|
}
|
|
_ = p.reportState(context.Background(), "reconnecting", nil)
|
|
p.mu.Lock()
|
|
p.lastReconnectAttemptAt = time.Now().UTC()
|
|
p.mu.Unlock()
|
|
if err := p.ensureConnected(); err != nil {
|
|
select {
|
|
case <-p.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
p.scheduleReconnect(err)
|
|
_ = p.reportState(context.Background(), "failed", err)
|
|
continue
|
|
}
|
|
select {
|
|
case <-p.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
p.resetReconnectState()
|
|
_ = p.reportState(context.Background(), "connected", nil)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
func connectionErrorCategory(err error) string {
|
|
if err == nil {
|
|
return ""
|
|
}
|
|
message := strings.ToLower(err.Error())
|
|
switch {
|
|
case strings.Contains(message, "auth"), strings.Contains(message, "password"), strings.Contains(message, "credential"):
|
|
return "authentication"
|
|
case strings.Contains(message, "heartbeat"):
|
|
return "heartbeat_timeout"
|
|
case strings.Contains(message, "timeout"):
|
|
return "timeout"
|
|
default:
|
|
return "network"
|
|
}
|
|
}
|
|
|
|
func reconnectDelay(attempt int, category string) time.Duration {
|
|
if category == "authentication" {
|
|
return defaultAuthReconnectDelay
|
|
}
|
|
if attempt < 1 {
|
|
attempt = 1
|
|
}
|
|
delays := []time.Duration{
|
|
defaultReconnectInitialDelay,
|
|
15 * time.Second,
|
|
30 * time.Second,
|
|
time.Minute,
|
|
2 * time.Minute,
|
|
defaultReconnectMaximumDelay,
|
|
}
|
|
delay := delays[min(attempt-1, len(delays)-1)]
|
|
// Deterministic ±10% jitter prevents a large set of channels from retrying together.
|
|
offsetPercent := (attempt*37)%21 - 10
|
|
delay += time.Duration(int64(delay) * int64(offsetPercent) / 100)
|
|
if delay < time.Second {
|
|
return time.Second
|
|
}
|
|
return delay
|
|
}
|
|
|
|
func isTemporaryReadTimeout(err error) bool {
|
|
var netErr net.Error
|
|
return errors.As(err, &netErr) && netErr.Timeout()
|
|
}
|