EvoBGP/internal/httpclient/circuit.go

package httpclient

import (
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"evobgp/internal/observability"
)

const (
	defaultBreakerThreshold = 5
	defaultBreakerCooldown  = 30 * time.Second
)

type hostBreaker struct {
	mu        sync.Mutex
	failures  int
	openUntil time.Time
}

// ErrBreakerCapacity rejects new upstream identities when the bounded cache is full.
var ErrBreakerCapacity = errors.New("httpclient: upstream circuit capacity exhausted")

type breakerEntry struct {
	breaker  *hostBreaker
	lastUsed time.Time
	active   int
}

type breakerCache struct {
	mu        sync.Mutex
	entries   map[string]*breakerEntry
	maximum   int
	idle      time.Duration
	nextSweep time.Time
}

var hostBreakers = breakerCache{entries: make(map[string]*breakerEntry), maximum: 1024, idle: 10 * time.Minute}

func (c *breakerCache) sweep(now time.Time) {
	for host, e := range c.entries {
		if e.active != 0 || now.Before(e.lastUsed.Add(c.idle)) {
			continue
		}
		e.breaker.mu.Lock()
		open := now.Before(e.breaker.openUntil)
		e.breaker.mu.Unlock()
		if !open {
			delete(c.entries, host)
			observability.RemoveUpstreamBreaker(host)
		}
	}
	c.nextSweep = now.Add(time.Second)
}

func (c *breakerCache) acquire(host string, now time.Time) (*hostBreaker, string, func(), error) {
	if len(host) > 254 {
		return nil, "", nil, fmt.Errorf("httpclient: upstream hostname too long")
	}
	host = strings.TrimSuffix(strings.ToLower(host), ".")
	if len(host) > 253 {
		return nil, "", nil, fmt.Errorf("httpclient: upstream hostname too long")
	}
	if host == "" {
		host = "_"
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if !now.Before(c.nextSweep) {
		c.sweep(now)
	}
	e := c.entries[host]
	if e == nil {
		if len(c.entries) >= c.maximum {
			return nil, host, nil, ErrBreakerCapacity
		}
		e = &breakerEntry{breaker: &hostBreaker{}}
		c.entries[host] = e
	}
	e.active++
	e.lastUsed = now
	var once sync.Once
	release := func() {
		once.Do(func() {
			c.mu.Lock()
			defer c.mu.Unlock()
			e.active--
			e.lastUsed = time.Now()
		})
	}
	return e.breaker, host, release, nil
}

func (b *hostBreaker) allow() bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	return time.Now().After(b.openUntil)
}

func (b *hostBreaker) recordSuccess() {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.failures = 0
	b.openUntil = time.Time{}
}

func (b *hostBreaker) recordFailure() {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.failures++
	if b.failures >= defaultBreakerThreshold {
		b.openUntil = time.Now().Add(defaultBreakerCooldown)
		b.failures = 0
	}
}

// SnapshotBreakers returns whether each known host breaker is currently open.
func SnapshotBreakers() map[string]bool {
	hostBreakers.mu.Lock()
	defer hostBreakers.mu.Unlock()
	now := time.Now()
	hostBreakers.sweep(now)
	out := map[string]bool{}
	for host, e := range hostBreakers.entries {
		b := e.breaker
		b.mu.Lock()
		open := time.Now().Before(b.openUntil)
		b.mu.Unlock()
		out[host] = open
	}
	return out
}

// ResetHostBreakers clears idle circuit breakers (tests only). In-flight
// references are retained so their completions cannot recreate orphan metrics.
func ResetHostBreakers() {
	hostBreakers.mu.Lock()
	defer hostBreakers.mu.Unlock()
	for host, e := range hostBreakers.entries {
		if e.active == 0 {
			delete(hostBreakers.entries, host)
			observability.RemoveUpstreamBreaker(host)
		}
	}
	hostBreakers.nextSweep = time.Time{}
}