openserp/core/circuit_breaker.go

package core

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type CircuitState int

const (
	CircuitClosed CircuitState = iota
	CircuitOpen
	CircuitHalfOpen
)

func (s CircuitState) String() string {
	switch s {
	case CircuitClosed:
		return "closed"
	case CircuitOpen:
		return "open"
	case CircuitHalfOpen:
		return "half-open"
	default:
		return "unknown"
	}
}

type CircuitBreakerConfig struct {
	FailureThreshold int
	RecoveryTimeout  time.Duration
	SuccessThreshold int
}

func DefaultCircuitBreakerConfig() CircuitBreakerConfig {
	return CircuitBreakerConfig{
		FailureThreshold: 5,
		RecoveryTimeout:  60 * time.Second,
		SuccessThreshold: 2,
	}
}

// CircuitBreaker tracks failure state for one engine.
type CircuitBreaker struct {
	mu              sync.RWMutex
	name            string
	state           CircuitState
	config          CircuitBreakerConfig
	failureCount    int
	successCount    int
	successLatency  time.Duration
	successSamples  int64
	lastFailureTime time.Time
	lastStateChange time.Time
}

func NewCircuitBreaker(name string, cfg CircuitBreakerConfig) *CircuitBreaker {
	return &CircuitBreaker{
		name:            name,
		state:           CircuitClosed,
		config:          cfg,
		lastStateChange: time.Now(),
	}
}

func (cb *CircuitBreaker) AllowRequest(ctx context.Context) bool {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	switch cb.state {
	case CircuitClosed:
		return true
	case CircuitOpen:
		if time.Since(cb.lastFailureTime) >= cb.config.RecoveryTimeout {
			cb.setState(CircuitHalfOpen)
			WithRequestEngine(ctx, cb.name).Info("Recovery timeout elapsed, moving to half-open")
			return true
		}
		return false
	case CircuitHalfOpen:
		return true
	default:
		return true
	}
}

func (cb *CircuitBreaker) RecordSuccess(ctx context.Context) {
	cb.RecordSuccessDuration(ctx, 0)
}

func (cb *CircuitBreaker) RecordSuccessDuration(ctx context.Context, elapsed time.Duration) {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	if elapsed > 0 {
		cb.successLatency += elapsed
		cb.successSamples++
	}

	switch cb.state {
	case CircuitHalfOpen:
		cb.successCount++
		if cb.successCount >= cb.config.SuccessThreshold {
			cb.setState(CircuitClosed)
			cb.failureCount = 0
			cb.successCount = 0
			WithRequestEngine(ctx, cb.name).Info("Circuit recovered, closed")
		}
	case CircuitClosed:
		cb.failureCount = 0
	}
}

func (cb *CircuitBreaker) RecordFailure(ctx context.Context) {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	cb.lastFailureTime = time.Now()

	switch cb.state {
	case CircuitClosed:
		cb.failureCount++
		if cb.failureCount >= cb.config.FailureThreshold {
			cb.setState(CircuitOpen)
			WithRequestEngine(ctx, cb.name).
				WithField("failure_count", cb.failureCount).
				WithField("recovery_timeout", cb.config.RecoveryTimeout.String()).
				Warn("Circuit opened after consecutive failures")
		}
	case CircuitHalfOpen:
		cb.setState(CircuitOpen)
		cb.successCount = 0
		WithRequestEngine(ctx, cb.name).Warn("Failed during half-open, circuit re-opened")
	}
}

func (cb *CircuitBreaker) State() CircuitState {
	cb.mu.RLock()
	defer cb.mu.RUnlock()
	return cb.state
}

func (cb *CircuitBreaker) Stats() map[string]interface{} {
	cb.mu.RLock()
	defer cb.mu.RUnlock()

	stats := map[string]interface{}{
		"engine":        cb.name,
		"state":         cb.state.String(),
		"failure_count": cb.failureCount,
		"last_changed":  cb.lastStateChange.Format(time.RFC3339),
	}

	if cb.state == CircuitOpen {
		remaining := cb.config.RecoveryTimeout - time.Since(cb.lastFailureTime)
		if remaining < 0 {
			remaining = 0
		}

		// Expose retry_in as integer seconds for easier client-side processing.
		retryInSeconds := int64(0)
		if remaining > 0 {
			retryInSeconds = int64((remaining + time.Second - time.Nanosecond) / time.Second)
		}
		stats["retry_in"] = retryInSeconds
	}
	if cb.successSamples > 0 {
		stats["avg_response_ms"] = int64((cb.successLatency / time.Duration(cb.successSamples)) / time.Millisecond)
	}

	return stats
}

func (cb *CircuitBreaker) AvgSuccessLatency() (time.Duration, bool) {
	cb.mu.RLock()
	defer cb.mu.RUnlock()
	if cb.successSamples == 0 {
		return 0, false
	}
	return cb.successLatency / time.Duration(cb.successSamples), true
}

func (cb *CircuitBreaker) setState(state CircuitState) {
	cb.state = state
	cb.lastStateChange = time.Now()
}

type CircuitBreakerManager struct {
	mu       sync.RWMutex
	breakers map[string]*CircuitBreaker
	config   CircuitBreakerConfig
}

func NewCircuitBreakerManager(cfg CircuitBreakerConfig) *CircuitBreakerManager {
	return &CircuitBreakerManager{
		breakers: make(map[string]*CircuitBreaker),
		config:   cfg,
	}
}

func (m *CircuitBreakerManager) Get(engineName string) *CircuitBreaker {
	m.mu.RLock()
	if cb, ok := m.breakers[engineName]; ok {
		m.mu.RUnlock()
		return cb
	}
	m.mu.RUnlock()

	m.mu.Lock()
	defer m.mu.Unlock()

	if cb, ok := m.breakers[engineName]; ok {
		return cb
	}

	cb := NewCircuitBreaker(engineName, m.config)
	m.breakers[engineName] = cb
	return cb
}

func (m *CircuitBreakerManager) AllStats() []map[string]interface{} {
	m.mu.RLock()
	defer m.mu.RUnlock()

	stats := make([]map[string]interface{}, 0, len(m.breakers))
	for _, cb := range m.breakers {
		stats = append(stats, cb.Stats())
	}
	return stats
}

var ErrCircuitOpen = fmt.Errorf("circuit breaker is open - engine temporarily disabled")