Fix sentinel error aliasing, hot-path allocations, and resource leaks

- Deduplicate sentinel errors: httpx.ErrNoHealthy, ErrCircuitOpen, and
  ErrRetryExhausted are now aliases to the canonical sub-package values
  so errors.Is works across package boundaries
- Retry transport returns ErrRetryExhausted only when all attempts are
  actually exhausted, not on early policy exit
- Balancer: pre-parse endpoint URLs at construction, replace req.Clone
  with cheap shallow struct copy to avoid per-request allocations
- Circuit breaker: Load before LoadOrStore to avoid allocating a Breaker
  on every request for known hosts
- Health checker: drain response body before close for connection reuse,
  probe endpoints concurrently, run initial probe synchronously in Start
- Client: add Close() to shut down health checker goroutine, propagate
  URL resolution errors instead of silently discarding them
- MockClock: fix lock ordering in Reset (clock.mu before t.mu), fix
  timer slice compaction to avoid backing-array aliasing, extract
  fireExpired to deduplicate Advance/Set
This commit is contained in:
2026-03-20 15:21:32 +03:00
parent f9a05f5c57
commit 5cfd1a7400
8 changed files with 155 additions and 48 deletions

View File

@@ -2,6 +2,7 @@ package balancer
import (
"errors"
"fmt"
"net/http"
"net/url"
@@ -23,6 +24,19 @@ type Strategy interface {
Next(healthy []Endpoint) (Endpoint, error)
}
// Closer can be used to shut down resources associated with a balancer
// transport (e.g. background health checker goroutines).
type Closer struct {
healthChecker *HealthChecker
}
// Close stops background goroutines. Safe to call multiple times.
func (c *Closer) Close() {
if c.healthChecker != nil {
c.healthChecker.Stop()
}
}
// Transport returns a middleware that load-balances requests across the
// provided endpoints using the configured strategy.
//
@@ -33,7 +47,7 @@ type Strategy interface {
// If active health checking is enabled (WithHealthCheck), a background
// goroutine periodically probes endpoints. Otherwise all endpoints are
// assumed healthy.
func Transport(endpoints []Endpoint, opts ...Option) middleware.Middleware {
func Transport(endpoints []Endpoint, opts ...Option) (middleware.Middleware, *Closer) {
o := &options{
strategy: RoundRobin(),
}
@@ -41,10 +55,22 @@ func Transport(endpoints []Endpoint, opts ...Option) middleware.Middleware {
opt(o)
}
// Pre-parse endpoint URLs once at construction time.
parsed := make(map[string]*url.URL, len(endpoints))
for _, ep := range endpoints {
u, err := url.Parse(ep.URL)
if err != nil {
panic(fmt.Sprintf("balancer: invalid endpoint URL %q: %v", ep.URL, err))
}
parsed[ep.URL] = u
}
if o.healthChecker != nil {
o.healthChecker.Start(endpoints)
}
closer := &Closer{healthChecker: o.healthChecker}
return func(next http.RoundTripper) http.RoundTripper {
return middleware.RoundTripperFunc(func(req *http.Request) (*http.Response, error) {
healthy := endpoints
@@ -61,18 +87,18 @@ func Transport(endpoints []Endpoint, opts ...Option) middleware.Middleware {
return nil, err
}
epURL, err := url.Parse(ep.URL)
if err != nil {
return nil, err
}
epURL := parsed[ep.URL]
// Clone the request URL and replace scheme+host with the endpoint.
r := req.Clone(req.Context())
// Shallow-copy request and URL to avoid mutating the original,
// without the expense of req.Clone's deep header copy.
r := *req
u := *req.URL
r.URL = &u
r.URL.Scheme = epURL.Scheme
r.URL.Host = epURL.Host
r.Host = epURL.Host
return next.RoundTrip(r)
return next.RoundTrip(&r)
})
}
}, closer
}

View File

@@ -33,7 +33,8 @@ func TestTransport_PicksEndpointAndReplacesURL(t *testing.T) {
return okResponse(), nil
})
rt := Transport(endpoints)(base)
mw, _ := Transport(endpoints)
rt := mw(base)
req, err := http.NewRequest(http.MethodGet, "https://original.example.com/api/v1/users", nil)
if err != nil {
@@ -67,7 +68,8 @@ func TestTransport_ErrNoHealthyWhenNoEndpoints(t *testing.T) {
return nil, nil
})
rt := Transport(endpoints)(base)
mw, _ := Transport(endpoints)
rt := mw(base)
req, err := http.NewRequest(http.MethodGet, "https://example.com/test", nil)
if err != nil {

View File

@@ -2,6 +2,7 @@ package balancer
import (
"context"
"io"
"net/http"
"sync"
"time"
@@ -70,8 +71,10 @@ func newHealthChecker(opts ...HealthOption) *HealthChecker {
}
// Start begins the background health checking loop for the given endpoints.
// All endpoints are initially considered healthy.
// An initial probe is run synchronously so that unhealthy endpoints are
// detected before the first request.
func (h *HealthChecker) Start(endpoints []Endpoint) {
// Mark all healthy as a safe default, then immediately probe.
h.mu.Lock()
for _, ep := range endpoints {
h.status[ep.URL] = true
@@ -82,6 +85,9 @@ func (h *HealthChecker) Start(endpoints []Endpoint) {
h.cancel = cancel
h.stopped = make(chan struct{})
// Run initial probe synchronously so callers don't hit stale state.
h.probe(ctx, endpoints)
go h.loop(ctx, endpoints)
}
@@ -111,7 +117,7 @@ func (h *HealthChecker) Healthy(endpoints []Endpoint) []Endpoint {
h.mu.RLock()
defer h.mu.RUnlock()
var result []Endpoint
result := make([]Endpoint, 0, len(endpoints))
for _, ep := range endpoints {
if h.status[ep.URL] {
result = append(result, ep)
@@ -137,13 +143,18 @@ func (h *HealthChecker) loop(ctx context.Context, endpoints []Endpoint) {
}
func (h *HealthChecker) probe(ctx context.Context, endpoints []Endpoint) {
var wg sync.WaitGroup
wg.Add(len(endpoints))
for _, ep := range endpoints {
healthy := h.check(ctx, ep)
h.mu.Lock()
h.status[ep.URL] = healthy
h.mu.Unlock()
go func() {
defer wg.Done()
healthy := h.check(ctx, ep)
h.mu.Lock()
h.status[ep.URL] = healthy
h.mu.Unlock()
}()
}
wg.Wait()
}
func (h *HealthChecker) check(ctx context.Context, ep Endpoint) bool {
@@ -156,6 +167,7 @@ func (h *HealthChecker) check(ctx context.Context, ep Endpoint) bool {
if err != nil {
return false
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
return resp.StatusCode >= 200 && resp.StatusCode < 300