sundynix-agentix/sundynix-dispatcher/internal/harness/circuitbreaker.go

// Package harness 提供 dispatcher 的治理组件（熔断降级 / 评测等）。
package harness

import (
	"log"
	"sync"
	"time"
)

// State 是熔断器状态。
type State int

const (
	Closed   State = iota // 闭合：正常放行
	Open                  // 断开：连续失败超阈值，快速拒绝
	HalfOpen              // 半开：冷却后放行少量探测，成功则恢复
)

func (s State) String() string {
	switch s {
	case Open:
		return "open"
	case HalfOpen:
		return "half-open"
	default:
		return "closed"
	}
}

// 默认参数。
const (
	defaultThreshold   = 5                // 闭合态连续失败达此数 → 断开
	defaultCooldown    = 10 * time.Second // 断开后多久转半开
	defaultHalfOpenMax = 1                // 半开态最多放行的探测数
)

// CircuitBreaker 实现熔断降级中心：后端连续失败时断开、快速拒绝，冷却后半开探测，
// 探测成功则恢复闭合、失败则重新断开。并发安全（多任务 goroutine 共享一个实例）。
type CircuitBreaker struct {
	mu             sync.Mutex
	state          State
	fails          int       // 闭合态连续失败计数
	openUntil      time.Time // 断开持续到的时间点
	halfOpenProbes int       // 半开态已放行的探测数

	threshold   int
	cooldown    time.Duration
	halfOpenMax int
	now         func() time.Time // 可注入时钟（测试用）
}

func NewCircuitBreaker() *CircuitBreaker {
	return NewCircuitBreakerWith(defaultThreshold, defaultCooldown, defaultHalfOpenMax)
}

// NewCircuitBreakerWith 用自定义参数建熔断器（非法入参回退默认）。
// 供模型层 failover 用更紧的阈值/冷却（比编排层更快跳过挂掉的主模型）；也便于测试注短冷却。
func NewCircuitBreakerWith(threshold int, cooldown time.Duration, halfOpenMax int) *CircuitBreaker {
	if threshold <= 0 {
		threshold = defaultThreshold
	}
	if cooldown <= 0 {
		cooldown = defaultCooldown
	}
	if halfOpenMax <= 0 {
		halfOpenMax = defaultHalfOpenMax
	}
	return &CircuitBreaker{
		state:       Closed,
		threshold:   threshold,
		cooldown:    cooldown,
		halfOpenMax: halfOpenMax,
		now:         time.Now,
	}
}

// Allow 判定当前是否放行请求，并在冷却到点时把断开切换为半开。
func (c *CircuitBreaker) Allow() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	switch c.state {
	case Open:
		if !c.now().Before(c.openUntil) { // 冷却到点 → 转半开，放行首个探测
			c.state = HalfOpen
			c.halfOpenProbes = 1
			log.Printf("[harness] 熔断器 open → half-open（放行探测）")
			return true
		}
		return false
	case HalfOpen:
		if c.halfOpenProbes < c.halfOpenMax {
			c.halfOpenProbes++
			return true
		}
		return false // 探测名额用尽，待 Report 决出结果
	default: // Closed
		return true
	}
}

// Report 上报一次调用结果以驱动状态机。
func (c *CircuitBreaker) Report(success bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if success {
		switch c.state {
		case HalfOpen:
			c.state = Closed
			c.fails = 0
			c.halfOpenProbes = 0
			log.Printf("[harness] 熔断器 half-open → closed（已恢复）")
		default:
			c.fails = 0
		}
		return
	}
	switch c.state {
	case HalfOpen:
		c.trip() // 探测失败 → 重新断开
	case Closed:
		c.fails++
		if c.fails >= c.threshold {
			c.trip()
		}
	}
}

// trip 切换到断开态并设定冷却到点。调用方须持锁。
func (c *CircuitBreaker) trip() {
	c.state = Open
	c.openUntil = c.now().Add(c.cooldown)
	c.halfOpenProbes = 0
	log.Printf("[harness] 熔断器断开（连续失败），%.0fs 后转半开", c.cooldown.Seconds())
}

// State 返回当前状态（观测 / 测试用）。
func (c *CircuitBreaker) State() State {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.state
}