broad-announce/internal/circuitbreaker/circuitbreaker.go

// Package circuitbreaker provides a per-component circuit breaker
// for the ingest pipeline (SPEC §22 layer 6). It wraps a function
// that performs an outbound call and trips the circuit when error
// rates exceed a threshold.
//
// State machine (3 states, 4 transitions):
//
//	CLOSED → OPEN   failure_count >= threshold within window
//	OPEN   → HALF   open_duration elapsed
//	HALF   → CLOSED successful call observed
//	HALF   → OPEN   call in HALF state also fails
//
// The breaker is in-process (no shared state across multiple
// ingestd instances). For the dev stack that is fine; for a
// multi-instance production deploy, the state would live in
// Redis (deferred to a future milestone).
//
// Metrics: callers are expected to call cb.Measure(state, err)
// after each Do() call so the gauge maintained here stays in
// sync with what the Prometheus scrape sees.
package circuitbreaker

import (
	"context"
	"errors"
	"sync"
	"time"
)

// State values. Mirrors the ba_ingestd_circuit_breaker_state gauge.
const (
	StateClosed   = 0
	StateHalfOpen = 1
	StateOpen     = 2
)

// Config is the static configuration for one circuit breaker.
// All fields must be set by the caller.
type Config struct {
	// Name is the component label used in Prometheus metrics.
	Name string
	// FailureThreshold is the number of consecutive failures
	// (within FailureWindow) that trips the circuit to OPEN.
	FailureThreshold int
	// FailureWindow is the rolling window for counting failures.
	FailureWindow time.Duration
	// OpenDuration is how long the circuit stays OPEN before
	// transitioning to HALF-OPEN (testing).
	OpenDuration time.Duration
	// MaxHalfOpen is the number of test calls admitted while
	// in HALF-OPEN state. Default 1 (admit one, decide).
	MaxHalfOpen int
}

// DefaultConfig is a reasonable starting point for a NATS broker
// circuit breaker: 5 failures in 10s trips; 30s open; 1 test call.
func DefaultConfig(name string) Config {
	return Config{
		Name:             name,
		FailureThreshold: 5,
		FailureWindow:    10 * time.Second,
		OpenDuration:     30 * time.Second,
		MaxHalfOpen:      1,
	}
}

// Breaker is the per-component circuit breaker. It is safe for
// concurrent use by the HTTP, MQTT, and WebSocket ingest paths.
type Breaker struct {
	config Config

	mu sync.RWMutex
	// state is one of StateClosed, StateHalfOpen, StateOpen.
	state int
	// failures is a circular buffer of recent failure timestamps.
	// We keep it as a slice and prune anything older than
	// config.FailureWindow on every call.
	failures []time.Time
	// halfOpenCount is how many test calls have been admitted
	// in the current HALF-OPEN window. Resets to 0 on transition
	// out of HALF-OPEN.
	halfOpenCount int
	// openSince is when we entered the OPEN state. Used to
	// decide when to transition to HALF-OPEN.
	openSince time.Time

	// Measure is called after every Do() call with the observed
	// state and error. It is nil in production; tests can inject
	// a spy to observe state transitions without polling.
	Measure func(state int, err error)
}

// New creates a new circuit breaker from cfg.
func New(cfg Config) *Breaker {
	if cfg.MaxHalfOpen <= 0 {
		cfg.MaxHalfOpen = 1
	}
	return &Breaker{config: cfg, state: StateClosed}
}

// Do runs fn if the circuit is CLOSED or HALF-OPEN. Returns
// ErrCircuitOpen when the circuit is OPEN. Returns the error
// from fn on failure; nil on success.
//
// If the circuit trips OPEN, Do records the failure internally
// so the next caller gets ErrCircuitOpen immediately.
func (cb *Breaker) Do(ctx context.Context, fn func() error) error {
	// Check context cancellation before doing any work.
	if err := ctx.Err(); err != nil {
		return err
	}

	// Check and handle any time-based transition BEFORE acquiring
	// the lock to avoid holding the lock across a time.Sleep.
	// (time.Since inside a mutex is a deadlock risk in writer-
	// biased RWMutex implementations).
	cb.tryHalfOpen()

	cb.mu.Lock()
	defer cb.mu.Unlock()

	switch cb.state {
	case StateOpen:
		return ErrCircuitOpen
	case StateHalfOpen:
		if cb.halfOpenCount >= cb.config.MaxHalfOpen {
			return ErrCircuitOpen
		}
		cb.halfOpenCount++
	}

	// Run the protected function.
	err := fn()

	// Record the result under the lock.
	cb.recordResultLocked(err)

	return err
}

// tryHalfOpen checks if an OPEN circuit's duration has elapsed
// and transitions it to HALF-OPEN. Safe to call without the lock;
// it acquires the lock internally for the write.
func (cb *Breaker) tryHalfOpen() {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	if cb.state == StateOpen && time.Since(cb.openSince) >= cb.config.OpenDuration {
		cb.state = StateHalfOpen
		cb.halfOpenCount = 0
		cb.failures = nil
		if cb.Measure != nil {
			cb.Measure(StateHalfOpen, nil)
		}
	}
}

// recordResultLocked updates internal state based on fn's result.
// Caller MUST hold cb.mu. Exported as recordResult for tests
// that hold the lock externally.
func (cb *Breaker) recordResultLocked(err error) {
	if err == nil {
		// Successful call.
		if cb.state == StateHalfOpen {
			cb.state = StateClosed
			cb.halfOpenCount = 0
			cb.failures = nil
			if cb.Measure != nil {
				cb.Measure(StateClosed, nil)
			}
		}
		return
	}

	// Failure.
	if cb.state == StateHalfOpen {
		// A failure in HALF-OPEN trips back to OPEN.
		cb.state = StateOpen
		cb.openSince = time.Now()
		cb.halfOpenCount = 0
		if cb.Measure != nil {
			cb.Measure(StateOpen, err)
		}
		return
	}

	// Failure in CLOSED: record it and check threshold.
	now := time.Now()
	cb.failures = append(cb.failures, now)

	// Prune anything outside the failure window.
	cutoff := now.Add(-cb.config.FailureWindow)
	j := 0
	for i, t := range cb.failures {
		if t.After(cutoff) {
			j = i
			break
		}
	}
	if j > 0 {
		cb.failures = cb.failures[j:]
	}

	// Check threshold.
	if len(cb.failures) >= cb.config.FailureThreshold {
		cb.state = StateOpen
		cb.openSince = now
		if cb.Measure != nil {
			cb.Measure(StateOpen, err)
		}
	}
}

// recordResult is a convenience wrapper for tests that don't
// already hold the lock.
func (cb *Breaker) recordResult(err error) {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	cb.recordResultLocked(err)
}

// State returns the current state (0=closed, 1=half-open, 2=open)
// without acquiring the write lock. Suitable for metrics reporting.
func (cb *Breaker) State() int {
	cb.mu.RLock()
	defer cb.mu.RUnlock()
	return cb.state
}

// ErrCircuitOpen is returned by Do when the circuit is OPEN.
var ErrCircuitOpen = errors.New("circuit breaker open")

// Ensure errors don't get shadowed.
var _ = ErrCircuitOpen.Error