configurable circuit breaker

Author: ndyakov

hitless/circuit_breaker.go

@@ -56,13 +56,16 @@ type CircuitBreaker struct {
 
 // newCircuitBreaker creates a new circuit breaker for an endpoint
 func newCircuitBreaker(endpoint string, config *Config) *CircuitBreaker {
-	// Use sensible defaults if not configured
-	failureThreshold := 10
-	resetTimeout := 500 * time.Millisecond
-	maxRequests := 10
-
-	// These could be added to Config in the future without breaking API
-	// For now, use internal defaults that work well
+	// Use configuration values with sensible defaults
+	failureThreshold := 5
+	resetTimeout := 60 * time.Second
+	maxRequests := 3
+
+	if config != nil {
+		failureThreshold = config.CircuitBreakerFailureThreshold
+		resetTimeout = config.CircuitBreakerResetTimeout
+		maxRequests = config.CircuitBreakerMaxRequests
+	}
 
 	return &CircuitBreaker{
 		failureThreshold: failureThreshold,

hitless/circuit_breaker_test.go

@@ -10,7 +10,10 @@ import (
 
 func TestCircuitBreaker(t *testing.T) {
 	config := &Config{
-		LogLevel: logging.LogLevelError, // Reduce noise in tests
+		LogLevel:                       logging.LogLevelError, // Reduce noise in tests
+		CircuitBreakerFailureThreshold: 5,
+		CircuitBreakerResetTimeout:     60 * time.Second,
+		CircuitBreakerMaxRequests:      3,
 	}
 
 	t.Run("InitialState", func(t *testing.T) {
@@ -44,7 +47,7 @@ func TestCircuitBreaker(t *testing.T) {
 	t.Run("FailureThreshold", func(t *testing.T) {
 		cb := newCircuitBreaker("test-endpoint:6379", config)
 		testError := errors.New("test error")
-		
+
 		// Fail 4 times (below threshold of 5)
 		for i := 0; i < 4; i++ {
 			err := cb.Execute(func() error {
@@ -57,15 +60,15 @@ func TestCircuitBreaker(t *testing.T) {
 				t.Errorf("Circuit should still be closed after %d failures", i+1)
 			}
 		}
-		
+
 		// 5th failure should open the circuit
 		err := cb.Execute(func() error {
 			return testError
 		})
 		if err != testError {
 			t.Errorf("Expected test error, got %v", err)
 		}
-		
+
 		if cb.GetState() != CircuitBreakerOpen {
 			t.Errorf("Expected state %v, got %v", CircuitBreakerOpen, cb.GetState())
 		}
@@ -92,8 +95,13 @@ func TestCircuitBreaker(t *testing.T) {
 	})
 
 	t.Run("HalfOpenTransition", func(t *testing.T) {
-		cb := newCircuitBreaker("test-endpoint:6379", config)
-		cb.resetTimeout = 100 * time.Millisecond // Short timeout for testing
+		testConfig := &Config{
+			LogLevel:                       logging.LogLevelError,
+			CircuitBreakerFailureThreshold: 5,
+			CircuitBreakerResetTimeout:     100 * time.Millisecond, // Short timeout for testing
+			CircuitBreakerMaxRequests:      3,
+		}
+		cb := newCircuitBreaker("test-endpoint:6379", testConfig)
 		testError := errors.New("test error")
 
 		// Force circuit to open
@@ -125,9 +133,13 @@ func TestCircuitBreaker(t *testing.T) {
 	})
 
 	t.Run("HalfOpenToClosedTransition", func(t *testing.T) {
-		cb := newCircuitBreaker("test-endpoint:6379", config)
-		cb.resetTimeout = 50 * time.Millisecond
-		cb.maxRequests = 3
+		testConfig := &Config{
+			LogLevel:                       logging.LogLevelError,
+			CircuitBreakerFailureThreshold: 5,
+			CircuitBreakerResetTimeout:     50 * time.Millisecond,
+			CircuitBreakerMaxRequests:      3,
+		}
+		cb := newCircuitBreaker("test-endpoint:6379", testConfig)
 		testError := errors.New("test error")
 
 		// Force circuit to open
@@ -155,8 +167,13 @@ func TestCircuitBreaker(t *testing.T) {
 	})
 
 	t.Run("HalfOpenToOpenOnFailure", func(t *testing.T) {
-		cb := newCircuitBreaker("test-endpoint:6379", config)
-		cb.resetTimeout = 50 * time.Millisecond
+		testConfig := &Config{
+			LogLevel:                       logging.LogLevelError,
+			CircuitBreakerFailureThreshold: 5,
+			CircuitBreakerResetTimeout:     50 * time.Millisecond,
+			CircuitBreakerMaxRequests:      3,
+		}
+		cb := newCircuitBreaker("test-endpoint:6379", testConfig)
 		testError := errors.New("test error")
 
 		// Force circuit to open
@@ -216,7 +233,10 @@ func TestCircuitBreaker(t *testing.T) {
 
 func TestCircuitBreakerManager(t *testing.T) {
 	config := &Config{
-		LogLevel: logging.LogLevelError,
+		LogLevel:                       logging.LogLevelError,
+		CircuitBreakerFailureThreshold: 5,
+		CircuitBreakerResetTimeout:     60 * time.Second,
+		CircuitBreakerMaxRequests:      3,
 	}
 
 	t.Run("GetCircuitBreaker", func(t *testing.T) {
@@ -289,4 +309,48 @@ func TestCircuitBreakerManager(t *testing.T) {
 			t.Error("Failure count should be reset to 0")
 		}
 	})
+
+	t.Run("ConfigurableParameters", func(t *testing.T) {
+		config := &Config{
+			LogLevel:                       logging.LogLevelError,
+			CircuitBreakerFailureThreshold: 10,
+			CircuitBreakerResetTimeout:     30 * time.Second,
+			CircuitBreakerMaxRequests:      5,
+		}
+
+		cb := newCircuitBreaker("test-endpoint:6379", config)
+
+		// Test that configuration values are used
+		if cb.failureThreshold != 10 {
+			t.Errorf("Expected failureThreshold=10, got %d", cb.failureThreshold)
+		}
+		if cb.resetTimeout != 30*time.Second {
+			t.Errorf("Expected resetTimeout=30s, got %v", cb.resetTimeout)
+		}
+		if cb.maxRequests != 5 {
+			t.Errorf("Expected maxRequests=5, got %d", cb.maxRequests)
+		}
+
+		// Test that circuit opens after configured threshold
+		testError := errors.New("test error")
+		for i := 0; i < 9; i++ {
+			err := cb.Execute(func() error { return testError })
+			if err != testError {
+				t.Errorf("Expected test error, got %v", err)
+			}
+			if cb.GetState() != CircuitBreakerClosed {
+				t.Errorf("Circuit should still be closed after %d failures", i+1)
+			}
+		}
+
+		// 10th failure should open the circuit
+		err := cb.Execute(func() error { return testError })
+		if err != testError {
+			t.Errorf("Expected test error, got %v", err)
+		}
+
+		if cb.GetState() != CircuitBreakerOpen {
+			t.Errorf("Expected state %v, got %v", CircuitBreakerOpen, cb.GetState())
+		}
+	})
 }

hitless/config.go

@@ -116,6 +116,19 @@ type Config struct {
 	// Default: logging.LogLevelError(0)
 	LogLevel logging.LogLevel
 
+	// Circuit breaker configuration for endpoint failure handling
+	// CircuitBreakerFailureThreshold is the number of failures before opening the circuit.
+	// Default: 5
+	CircuitBreakerFailureThreshold int
+
+	// CircuitBreakerResetTimeout is how long to wait before testing if the endpoint recovered.
+	// Default: 60 seconds
+	CircuitBreakerResetTimeout time.Duration
+
+	// CircuitBreakerMaxRequests is the maximum number of requests allowed in half-open state.
+	// Default: 3
+	CircuitBreakerMaxRequests int
+
 	// MaxHandoffRetries is the maximum number of times to retry a failed handoff.
 	// After this many retries, the connection will be removed from the pool.
 	// Default: 3
@@ -138,6 +151,11 @@ func DefaultConfig() *Config {
 		PostHandoffRelaxedDuration: 0, // Auto-calculated based on relaxed timeout
 		LogLevel:                   logging.LogLevelError,
 
+		// Circuit breaker configuration
+		CircuitBreakerFailureThreshold: 5,
+		CircuitBreakerResetTimeout:     60 * time.Second,
+		CircuitBreakerMaxRequests:      3,
+
 		// Connection Handoff Configuration
 		MaxHandoffRetries: 3,
 	}
@@ -167,6 +185,17 @@ func (c *Config) Validate() error {
 		return ErrInvalidLogLevel
 	}
 
+	// Circuit breaker validation
+	if c.CircuitBreakerFailureThreshold < 1 {
+		return ErrInvalidCircuitBreakerFailureThreshold
+	}
+	if c.CircuitBreakerResetTimeout < 0 {
+		return ErrInvalidCircuitBreakerResetTimeout
+	}
+	if c.CircuitBreakerMaxRequests < 1 {
+		return ErrInvalidCircuitBreakerMaxRequests
+	}
+
 	// Validate Mode (maintenance notifications mode)
 	if !c.Mode.IsValid() {
 		return ErrInvalidMaintNotifications
@@ -280,6 +309,22 @@ func (c *Config) ApplyDefaultsWithPoolConfig(poolSize int, maxActiveConns int) *
 		result.MaxHandoffRetries = c.MaxHandoffRetries
 	}
 
+	// Circuit breaker configuration
+	result.CircuitBreakerFailureThreshold = defaults.CircuitBreakerFailureThreshold
+	if c.CircuitBreakerFailureThreshold > 0 {
+		result.CircuitBreakerFailureThreshold = c.CircuitBreakerFailureThreshold
+	}
+
+	result.CircuitBreakerResetTimeout = defaults.CircuitBreakerResetTimeout
+	if c.CircuitBreakerResetTimeout > 0 {
+		result.CircuitBreakerResetTimeout = c.CircuitBreakerResetTimeout
+	}
+
+	result.CircuitBreakerMaxRequests = defaults.CircuitBreakerMaxRequests
+	if c.CircuitBreakerMaxRequests > 0 {
+		result.CircuitBreakerMaxRequests = c.CircuitBreakerMaxRequests
+	}
+
 	if result.LogLevel.DebugOrAbove() {
 		internal.Logger.Printf(context.Background(), "hitless: debug logging enabled")
 		internal.Logger.Printf(context.Background(), "hitless: config: %+v", result)
@@ -303,6 +348,11 @@ func (c *Config) Clone() *Config {
 		PostHandoffRelaxedDuration: c.PostHandoffRelaxedDuration,
 		LogLevel:                   c.LogLevel,
 
+		// Circuit breaker configuration
+		CircuitBreakerFailureThreshold: c.CircuitBreakerFailureThreshold,
+		CircuitBreakerResetTimeout:     c.CircuitBreakerResetTimeout,
+		CircuitBreakerMaxRequests:      c.CircuitBreakerMaxRequests,
+
 		// Configuration fields
 		MaxHandoffRetries: c.MaxHandoffRetries,
 	}

hitless/config_test.go

@@ -33,6 +33,17 @@ func TestConfig(t *testing.T) {
 			t.Errorf("Expected MaxHandoffRetries to be 3, got %d", config.MaxHandoffRetries)
 		}
 
+		// Circuit breaker defaults
+		if config.CircuitBreakerFailureThreshold != 5 {
+			t.Errorf("Expected CircuitBreakerFailureThreshold=5, got %d", config.CircuitBreakerFailureThreshold)
+		}
+		if config.CircuitBreakerResetTimeout != 60*time.Second {
+			t.Errorf("Expected CircuitBreakerResetTimeout=60s, got %v", config.CircuitBreakerResetTimeout)
+		}
+		if config.CircuitBreakerMaxRequests != 3 {
+			t.Errorf("Expected CircuitBreakerMaxRequests=3, got %d", config.CircuitBreakerMaxRequests)
+		}
+
 		if config.HandoffTimeout != 15*time.Second {
 			t.Errorf("Expected HandoffTimeout to be 15s, got %v", config.HandoffTimeout)
 		}
@@ -377,6 +388,25 @@ func TestEnhancedConfigValidation(t *testing.T) {
 		}
 		config.MaxHandoffRetries = 3 // Reset to valid value
 
+		// Test circuit breaker validation
+		config.CircuitBreakerFailureThreshold = 0
+		if err := config.Validate(); err != ErrInvalidCircuitBreakerFailureThreshold {
+			t.Errorf("Expected ErrInvalidCircuitBreakerFailureThreshold, got %v", err)
+		}
+		config.CircuitBreakerFailureThreshold = 5 // Reset to valid value
+
+		config.CircuitBreakerResetTimeout = -1 * time.Second
+		if err := config.Validate(); err != ErrInvalidCircuitBreakerResetTimeout {
+			t.Errorf("Expected ErrInvalidCircuitBreakerResetTimeout, got %v", err)
+		}
+		config.CircuitBreakerResetTimeout = 60 * time.Second // Reset to valid value
+
+		config.CircuitBreakerMaxRequests = 0
+		if err := config.Validate(); err != ErrInvalidCircuitBreakerMaxRequests {
+			t.Errorf("Expected ErrInvalidCircuitBreakerMaxRequests, got %v", err)
+		}
+		config.CircuitBreakerMaxRequests = 3 // Reset to valid value
+
 		// Should pass validation again
 		if err := config.Validate(); err != nil {
 			t.Errorf("Config should be valid after reset, got error: %v", err)

hitless/errors.go

@@ -53,3 +53,10 @@ var (
 var (
 	ErrCircuitBreakerOpen = errors.New("hitless: circuit breaker is open, failing fast")
 )
+
+// circuit breaker configuration errors
+var (
+	ErrInvalidCircuitBreakerFailureThreshold = errors.New("hitless: circuit breaker failure threshold must be >= 1")
+	ErrInvalidCircuitBreakerResetTimeout     = errors.New("hitless: circuit breaker reset timeout must be >= 0")
+	ErrInvalidCircuitBreakerMaxRequests      = errors.New("hitless: circuit breaker max requests must be >= 1")
+)

hitless/handoff_worker.go

@@ -44,7 +44,7 @@ func newHandoffWorkerManager(config *Config, poolHook *PoolHook) *handoffWorkerM
 		handoffQueue:          make(chan HandoffRequest, config.HandoffQueueSize),
 		shutdown:              make(chan struct{}),
 		maxWorkers:            config.MaxWorkers,
-		activeWorkers:         atomic.Int32{}, // Start with no workers - create on demand
+		activeWorkers:         atomic.Int32{},   // Start with no workers - create on demand
 		workerTimeout:         15 * time.Second, // Workers exit after 15s of inactivity
 		config:                config,
 		poolHook:              poolHook,
@@ -423,14 +423,14 @@ func (hwm *handoffWorkerManager) closeConnFromRequest(ctx context.Context, reque
 		pooler.Remove(ctx, conn, err)
 		if hwm.config != nil && hwm.config.LogLevel.WarnOrAbove() { // Warning level
 			internal.Logger.Printf(ctx,
-				"hitless: removed conn[%d] from pool due to max handoff retries reached: %v",
+				"hitless: removed conn[%d] from pool due: %v",
 				conn.GetID(), err)
 		}
 	} else {
 		conn.Close()
 		if hwm.config != nil && hwm.config.LogLevel.WarnOrAbove() { // Warning level
 			internal.Logger.Printf(ctx,
-				"hitless: no pool provided for conn[%d], cannot remove due to handoff initialization failure: %v",
+				"hitless: no pool provided for conn[%d], cannot remove due to: %v",
 				conn.GetID(), err)
 		}
 	}