31bf3e5907
CircuitBreaker 此前是空桩(Allow 恒 true、Report 空操作),dispatcher 调 LLM/工具 无任何失败保护——今天就撞上 DeepSeek 流连接累积把报告卡死。改为真实三态熔断: - Closed:正常放行;连续失败达阈值(默认5) → Open。 - Open:快速拒绝;冷却(默认10s)到点 → HalfOpen 放行少量探测(默认1)。 - HalfOpen:探测成功 → Closed 恢复;探测失败 → 重新 Open。 - sync.Mutex 并发安全(多任务 goroutine 共享);时钟可注入便于确定性测试。 orchestrator.Handle:熔断开启时不再静默丢弃任务,改为回流"服务繁忙"提示 + CompleteStream 收尾,让客户端解阻不挂死。 测试(含 -race):达阈值断开、成功清零、半开恢复、探测失败重断、并发安全 —— 全过。 PROGRESS.md 勾掉熔断项。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
127 lines
3.1 KiB
Go
127 lines
3.1 KiB
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 &CircuitBreaker{
|
|
state: Closed,
|
|
threshold: defaultThreshold,
|
|
cooldown: defaultCooldown,
|
|
halfOpenMax: defaultHalfOpenMax,
|
|
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
|
|
}
|