// Package resilience provides per-dependency circuit breaking and concurrency isolation. package resilience import ( "context" "errors" "sync" "time" ) var ErrOpen = errors.New("resilience: circuit open") var ErrBusy = errors.New("resilience: concurrency limit reached") type Config struct { FailureThreshold int `json:"failureThreshold"` MaxConcurrent int `json:"maxConcurrent"` OpenTimeout time.Duration `json:"openTimeout"` IsFailure func(error) bool `json:"-"` } type Breaker struct { mu sync.Mutex config Config active, failures int state string until time.Time generation uint64 now func() time.Time } type Snapshot struct { State string `json:"state"` Active int `json:"active"` Failures int `json:"failures"` } func New(cfg Config) (*Breaker, error) { if cfg.FailureThreshold < 0 || cfg.MaxConcurrent < 0 || cfg.OpenTimeout < 0 { return nil, errors.New("resilience: invalid configuration") } if cfg.FailureThreshold == 0 { cfg.FailureThreshold = 5 } if cfg.MaxConcurrent == 0 { cfg.MaxConcurrent = 32 } if cfg.OpenTimeout == 0 { cfg.OpenTimeout = 5 * time.Second } if cfg.IsFailure == nil { cfg.IsFailure = func(err error) bool { return err != nil && !errors.Is(err, context.Canceled) } } return &Breaker{config: cfg, state: "closed", now: time.Now}, nil } // Acquire returns a completion callback; it must be called exactly once. // A half-open circuit admits one probe. Old-generation results cannot close it. func (b *Breaker) Acquire(ctx context.Context) (func(bool), error) { if ctx == nil { return nil, errors.New("resilience: nil context") } if err := ctx.Err(); err != nil { return nil, err } b.mu.Lock() defer b.mu.Unlock() if b.state == "open" { if b.now().Before(b.until) { return nil, ErrOpen } b.state = "halfOpen" b.generation++ } if b.state == "halfOpen" && b.active > 0 { return nil, ErrOpen } if b.active >= b.config.MaxConcurrent { return nil, ErrBusy } b.active++ generation := b.generation var once sync.Once return func(success bool) { once.Do(func() { b.mu.Lock() defer b.mu.Unlock() b.active-- if generation != b.generation { return } if success { b.failures = 0 if b.state == "halfOpen" { b.state = "closed" b.generation++ } return } b.failures++ if b.state == "halfOpen" || b.failures >= b.config.FailureThreshold { b.state = "open" b.until = b.now().Add(b.config.OpenTimeout) b.generation++ } }) }, nil } func (b *Breaker) Do(ctx context.Context, work func(context.Context) error) (err error) { if work == nil { return errors.New("resilience: nil work") } done, err := b.Acquire(ctx) if err != nil { return err } complete := false defer func() { if !complete { done(false) } }() err = work(ctx) done(!b.config.IsFailure(err)) complete = true return err } func (b *Breaker) Snapshot() Snapshot { b.mu.Lock() defer b.mu.Unlock() return Snapshot{State: b.state, Active: b.active, Failures: b.failures} }