resiliency

func Retry(ctx context.Context, fn func() error, opts ...func(*RetryOptions)) error

{
	if fn == nil {
		return errors.New("resiliency: retry function cannot be nil")
	}
	o := DefaultRetryOptions
	for _, opt := range opts {
		opt(&o)
	}
	if err := validateRetryOptions(o); err != nil {
		return err
	}

	var lastErr error
	for i := 0; i < o.Attempts; i++ {
		if err := ctx.Err(); err != nil {
			return err
		}

		if err := fn(); err == nil {
			return nil
		} else {
			lastErr = err
			if o.RetryIf != nil && !o.RetryIf(err) {
				return err
			}
		}

		if i < o.Attempts-1 {
			delay := time.Duration(float64(o.InitialDelay) * math.Pow(o.Factor, float64(i)))
			if delay > o.MaxDelay {
				delay = o.MaxDelay
			}

			if o.Jitter > 0 {
				jitter := time.Duration(float64(delay) * o.Jitter * rand.Float64())
				delay += jitter
			}

			timer := time.NewTimer(delay)
			select {
			case <-ctx.Done():
				timer.Stop()
				return ctx.Err()
			case <-timer.C:
			}
		}
	}

	return lastErr
}

func validateRetryOptions(options RetryOptions) error

{
	switch {
	case options.Attempts < 1:
		return errors.New("resiliency: retry attempts must be at least 1")
	case options.InitialDelay < 0:
		return errors.New("resiliency: initial delay cannot be negative")
	case options.MaxDelay < options.InitialDelay:
		return errors.New("resiliency: max delay cannot be less than initial delay")
	case math.IsNaN(options.Factor) || math.IsInf(options.Factor, 0) || options.Factor < 1:
		return fmt.Errorf("resiliency: retry factor must be finite and at least 1")
	case math.IsNaN(options.Jitter) || math.IsInf(options.Jitter, 0) || options.Jitter < 0 || options.Jitter > 1:
		return errors.New("resiliency: jitter must be between 0 and 1")
	default:
		return nil
	}
}

func TestRetry(t *testing.T)

{
	t.Run("SuccessFirstTry", func(t *testing.T) {
		calls := 0
		err := Retry(context.Background(), func() error {
			calls++
			return nil
		})
		if err != nil || calls != 1 {
			t.Errorf("Retry failed: %v, calls=%d", err, calls)
		}
	})

	t.Run("SuccessAfterRetries", func(t *testing.T) {
		calls := 0
		err := Retry(context.Background(), func() error {
			calls++
			if calls < 3 {
				return errors.New("fail")
			}
			return nil
		}, WithAttempts(5), WithDelay(1*time.Millisecond, 10*time.Millisecond))

		if err != nil || calls != 3 {
			t.Errorf("Retry failed: %v, calls=%d", err, calls)
		}
	})

	t.Run("FailureAllAttempts", func(t *testing.T) {
		calls := 0
		targetErr := errors.New("permanent fail")
		err := Retry(context.Background(), func() error {
			calls++
			return targetErr
		}, WithAttempts(3), WithDelay(1*time.Millisecond, 1*time.Millisecond))

		if err != targetErr || calls != 3 {
			t.Errorf("Retry should return last error: %v, calls=%d", err, calls)
		}
	})

	t.Run("ContextCancellation", func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())
		cancel()

		calls := 0
		err := Retry(ctx, func() error {
			calls++
			return errors.New("fail")
		})

		if err != context.Canceled || calls != 0 {
			t.Errorf("Retry should stop on context cancel: %v", err)
		}
	})
}

func TestRetryRejectsInvalidOptions(t *testing.T)

{
	calls := 0
	fn := func() error {
		calls++
		return nil
	}
	tests := []func(*RetryOptions){
		WithAttempts(0),
		WithDelay(-time.Second, time.Second),
		WithDelay(time.Second, 0),
		WithFactor(0),
		WithJitter(2),
	}
	for _, option := range tests {
		if err := Retry(context.Background(), fn, option); err == nil {
			t.Fatal("Retry() accepted invalid options")
		}
	}
	if calls != 0 {
		t.Fatalf("Retry() called function %d times for invalid options", calls)
	}
}

func TestCircuitBreaker(t *testing.T)

{
	cb := NewCircuitBreaker(2, 50*time.Millisecond)

	// State: Closed
	err := cb.Execute(func() error { return nil })
	if err != nil || cb.State() != StateClosed {
		t.Error("Circuit should be closed and return nil")
	}

	// First failure
	_ = cb.Execute(func() error { return errors.New("fail1") })
	if cb.State() != StateClosed {
		t.Error("Circuit should still be closed after 1 failure")
	}

	// Second failure -> State: Open
	_ = cb.Execute(func() error { return errors.New("fail2") })
	if cb.State() != StateOpen {
		t.Error("Circuit should be open after reaching threshold")
	}

	// While Open
	err = cb.Execute(func() error { return nil })
	if err != ErrCircuitOpen {
		t.Errorf("Execute should return ErrCircuitOpen when open, got %v", err)
	}

	// Wait for timeout -> State: Half-Open (on next Execute)
	time.Sleep(60 * time.Millisecond)

	// First success in Half-Open -> State: Closed
	err = cb.Execute(func() error { return nil })
	if err != nil || cb.State() != StateClosed {
		t.Errorf("Circuit should be closed after success in half-open, got state %v", cb.State())
	}
}

func TestHalfOpenAllowsOnlyOneProbe(t *testing.T)

{
	const timeout = 30 * time.Millisecond
	cb := NewCircuitBreaker(1, timeout)

	// Trip the breaker.
	_ = cb.Execute(func() error { return errors.New("fail") })
	if cb.State() != StateOpen {
		t.Fatal("expected StateOpen after threshold reached")
	}

	// Wait for the open timeout to elapse so the next allow() transitions to HalfOpen.
	time.Sleep(timeout + 10*time.Millisecond)

	// Fire 5 concurrent requests; use a gate so they all call allow() at the same time.
	const n = 5
	var (
		wg      sync.WaitGroup
		gate    = make(chan struct{})
		allowed int32
		blocked int32
	)

	for i := 0; i < n; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-gate
			err := cb.Execute(func() error {
				// Slow probe so concurrent goroutines can observe HalfOpen.
				time.Sleep(20 * time.Millisecond)
				return nil
			})
			if err == ErrCircuitOpen {
				atomic.AddInt32(&blocked, 1)
			} else if err == nil {
				atomic.AddInt32(&allowed, 1)
			}
		}()
	}

	close(gate)
	wg.Wait()

	if allowed != 1 {
		t.Errorf("expected exactly 1 probe allowed, got %d", allowed)
	}
	if blocked != n-1 {
		t.Errorf("expected %d blocked, got %d", n-1, blocked)
	}
	if cb.State() != StateClosed {
		t.Errorf("expected StateClosed after successful probe, got %v", cb.State())
	}
}

func TestHalfOpenFailureResetsCounter(t *testing.T)

{
	const timeout = 30 * time.Millisecond
	cb := NewCircuitBreaker(2, timeout)

	// Reach threshold to open breaker (2 failures).
	_ = cb.Execute(func() error { return errors.New("f1") })
	_ = cb.Execute(func() error { return errors.New("f2") })
	if cb.State() != StateOpen {
		t.Fatal("expected StateOpen")
	}

	// The first HalfOpen probe fails and returns to Open.
	time.Sleep(timeout + 10*time.Millisecond)
	_ = cb.Execute(func() error { return errors.New("probe fail") })
	if cb.State() != StateOpen {
		t.Fatalf("expected StateOpen after half-open probe failure, got %v", cb.State())
	}

	// Second HalfOpen: failures counter must have been reset to 0 on the previous
	// A single failure in the new HalfOpen cycle should
	// send the breaker back to Open immediately (not leave it closed/half-open).
	time.Sleep(timeout + 10*time.Millisecond)
	_ = cb.Execute(func() error { return errors.New("probe fail 2") })
	if cb.State() != StateOpen {
		t.Errorf("expected StateOpen after second half-open failure, got %v", cb.State())
	}
}

func TestCircuitBreakerStateCallbackCanReenter(t *testing.T)

{
	cb := NewCircuitBreaker(1, time.Hour)
	callbackDone := make(chan struct{})
	cb.OnStateChange(func(_, to State) {
		if got := cb.State(); got != to {
			t.Errorf("State() = %v inside callback, want %v", got, to)
		}
		cb.OnStateChange(nil)
		close(callbackDone)
	})

	_ = cb.Execute(func() error { return errors.New("failure") })

	select {
	case <-callbackDone:
	case <-time.After(time.Second):
		t.Fatal("state callback deadlocked")
	}
}

func NewRateLimiter(rate, burst int) (*RateLimiter, error)

{
	if rate <= 0 {
		return nil, fmt.Errorf("resiliency: rate must be positive")
	}
	if burst <= 0 {
		return nil, fmt.Errorf("resiliency: burst must be positive")
	}
	return &RateLimiter{
		tokens:   float64(burst),
		rate:     float64(rate),
		burst:    float64(burst),
		lastTime: time.Now(),
	}, nil
}

func NewBulkhead(maxConcurrent, maxQueue int) (*Bulkhead, error)

{
	if maxConcurrent <= 0 {
		return nil, fmt.Errorf("resiliency: maxConcurrent must be positive")
	}
	if maxQueue < 0 {
		return nil, fmt.Errorf("resiliency: maxQueue cannot be negative")
	}
	return &Bulkhead{
		slots: make(chan struct{}, maxConcurrent),
		queue: make(chan struct{}, maxQueue),
	}, nil
}

func TestNewRateLimiterRejectsInvalidConfiguration(t *testing.T)

{
	for _, test := range []struct {
		rate  int
		burst int
	}{
		{rate: 0, burst: 1},
		{rate: 1, burst: 0},
		{rate: -1, burst: 1},
	} {
		if _, err := NewRateLimiter(test.rate, test.burst); err == nil {
			t.Fatalf("NewRateLimiter(%d, %d) accepted invalid values", test.rate, test.burst)
		}
	}
}

func TestBulkheadEnforcesConcurrencyAndQueueLimits(t *testing.T)

{
	bulkhead, err := NewBulkhead(1, 1)
	if err != nil {
		t.Fatal(err)
	}
	release := make(chan struct{})
	started := make(chan struct{}, 2)
	var running atomic.Int32
	var peak atomic.Int32
	run := func() error {
		current := running.Add(1)
		for {
			previous := peak.Load()
			if current <= previous || peak.CompareAndSwap(previous, current) {
				break
			}
		}
		started <- struct{}{}
		<-release
		running.Add(-1)
		return nil
	}

	first := make(chan error, 1)
	second := make(chan error, 1)
	go func() { first <- bulkhead.Execute(context.Background(), run) }()
	<-started
	go func() { second <- bulkhead.Execute(context.Background(), run) }()

	deadline := time.Now().Add(time.Second)
	for len(bulkhead.queue) != 1 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if err := bulkhead.Execute(context.Background(), run); !errors.Is(err, ErrBulkheadFull) {
		t.Fatalf("third Execute() error = %v, want ErrBulkheadFull", err)
	}

	release <- struct{}{}
	<-started
	release <- struct{}{}
	if err := <-first; err != nil {
		t.Fatal(err)
	}
	if err := <-second; err != nil {
		t.Fatal(err)
	}
	if got := peak.Load(); got != 1 {
		t.Fatalf("peak concurrency = %d, want 1", got)
	}
}