// 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