onesvm-browser-server/server/internal/proxymanager/health.go

// health.go：探活状态机（design §5.2 / proxy-probe §6.2–6.4）。
//
// 调度：活跃出口 30s/次、全池 5min/轮（并发 6、单节点 8s 超时，
// 经 mihomo delay API 逐节点测）；连续 2 次失败摘除、1 次成功回候选；
// EWMA 延迟按区域组维护（design §5.2「按域名分」首版降为按区域组，
// 回执注明降级）。
//
// 状态机：candidate →(连续2败)→ removed →(1胜)→ candidate；
// active 出口失败立即切热备（由 manager 层处理重选）。
package proxymanager

import (
	"context"
	"io"
	"log"
	"math"
	"sort"
	"sync"
	"time"

	"onesvm.com/onesvm/browser-server/internal/config"
)

// 状态机参数（design §5.2 锁定值）。
const (
	activeInterval   = 30 * time.Second // 活跃出口探活频率
	fullPoolInterval = 5 * time.Minute  // 全池轮询频率
	probeConcurrency = 6                // 全池并发
	failThreshold    = 2                // 连续失败摘除阈值
	ewmaAlpha        = 0.3              // EWMA 平滑系数（新样本权重）
	ewmaInitSamples  = 2                // 初始样本数后 EWMA 收敛
)

// 节点状态。
const (
	stateCandidate = "candidate" // 候选（可被 P2C 选中）
	stateRemoved   = "removed"   // 摘除（连续 2 败）
)

// nodeState 单节点探活状态。
type nodeState struct {
	Name        string
	Region      string
	Pool        string
	State       string // candidate | removed
	ConsecFails int    // 连续失败计数
	LastDelayMs int    // 最近一次延迟（0=从未成功）
	LastProbeAt time.Time
}

// healthStat EWMA 统计（按区域组聚合维度之一）。
type healthStat struct {
	ewma    float64 // 区域组内平均延迟 EWMA
	samples int
}

// HealthEngine 探活状态机（并发安全）。
type HealthEngine struct {
	mu       sync.RWMutex
	nodes    map[string]*nodeState  // key=node name
	byRegion map[string]*healthStat // EWMA 按区域组（首版降级，回执注明）
	prober   Prober
	logger   *log.Logger
}

// Prober 探针接口（生产= mihomo delay API；测试= stub）。
type Prober interface {
	ProbeNode(ctx context.Context, name string) (delayMs int, err error)
}

// NewHealthEngine 构造探活引擎。
func NewHealthEngine(prober Prober, logger *log.Logger) *HealthEngine {
	if logger == nil {
		logger = log.New(io.Discard, "proxymanager/health ", 0)
	}
	return &HealthEngine{
		nodes:    map[string]*nodeState{},
		byRegion: map[string]*healthStat{},
		prober:   prober,
		logger:   logger,
	}
}

// ReplacePool 全量替换节点集（订阅重载后调用；保留同名节点既有状态机进度）。
func (h *HealthEngine) ReplacePool(metas []ProxyMeta) {
	h.mu.Lock()
	defer h.mu.Unlock()
	next := make(map[string]*nodeState, len(metas))
	for _, m := range metas {
		if m.Pool != poolVless {
			continue // 探活只覆盖可调度池
		}
		if old, ok := h.nodes[m.Name]; ok {
			old.Region, old.Pool = m.Region, m.Pool
			next[m.Name] = old
			continue
		}
		next[m.Name] = &nodeState{Name: m.Name, Region: m.Region, Pool: m.Pool, State: stateCandidate}
	}
	h.nodes = next
	// 区域统计重建（保留仍存在区域的历史 EWMA）。
	nextRegion := map[string]*healthStat{}
	for r := range h.byRegion {
		for _, n := range next {
			if n.Region == r {
				nextRegion[r] = h.byRegion[r]
				break
			}
		}
	}
	h.byRegion = nextRegion
}

// probeOne 探测单节点并推进状态机。返回本次是否成功。
func (h *HealthEngine) probeOne(ctx context.Context, n *nodeState) bool {
	if h.prober == nil || n == nil {
		return false
	}
	delay, err := h.prober.ProbeNode(ctx, n.Name)
	h.mu.Lock()
	defer h.mu.Unlock()
	n.LastProbeAt = config.Now()
	if err != nil {
		n.ConsecFails++
		if n.ConsecFails >= failThreshold && n.State == stateCandidate {
			n.State = stateRemoved
			h.logger.Printf("[health] 节点摘除 name=%s region=%s consec_fails=%d", n.Name, n.Region, n.ConsecFails)
		}
		return false
	}
	wasRemoved := n.State == stateRemoved
	n.ConsecFails = 0
	n.LastDelayMs = delay
	n.State = stateCandidate // 1 次成功即回候选
	if wasRemoved {
		h.logger.Printf("[health] 节点恢复 name=%s region=%s delay_ms=%d", n.Name, n.Region, delay)
	}
	// EWMA 按区域组更新。
	st := h.byRegion[n.Region]
	if st == nil {
		st = &healthStat{}
		h.byRegion[n.Region] = st
	}
	if st.samples == 0 {
		st.ewma = float64(delay)
	} else {
		st.ewma = ewmaAlpha*float64(delay) + (1-ewmaAlpha)*st.ewma
	}
	st.samples++
	return true
}

// ProbeActive 探活活跃出口集合（30s 周期调用；单并发，延迟敏感）。
// 返回 (全部成功?, 失败节点名列表)。
func (h *HealthEngine) ProbeActive(ctx context.Context, activeNames []string) (bool, []string) {
	var failed []string
	for _, name := range activeNames {
		h.mu.RLock()
		n := h.nodes[name]
		h.mu.RUnlock()
		if n == nil {
			continue
		}
		if !h.probeOne(ctx, n) {
			failed = append(failed, name)
		}
	}
	return len(failed) == 0, failed
}

// ProbeFullPool 全池轮询（5min 周期，并发 6）。返回 (候选数, 摘除数)。
func (h *HealthEngine) ProbeFullPool(ctx context.Context) (int, int) {
	h.mu.RLock()
	names := make([]*nodeState, 0, len(h.nodes))
	for _, n := range h.nodes {
		names = append(names, n)
	}
	h.mu.RUnlock()
	var (
		wg      sync.WaitGroup
		sem     = make(chan struct{}, probeConcurrency)
		mu      sync.Mutex
		alive   int
		removed int
	)
	for _, n := range names {
		wg.Add(1)
		go func(n *nodeState) {
			defer wg.Done()
			sem <- struct{}{}
			defer func() { <-sem }()
			if h.probeOne(ctx, n) {
				mu.Lock()
				alive++
				mu.Unlock()
			} else {
				mu.Lock()
				if h.snapshotState(n.Name) == stateRemoved {
					removed++
				}
				mu.Unlock()
			}
		}(n)
	}
	wg.Wait()
	return alive, removed
}

// snapshotState 读取单节点状态（ProbeFullPool 内部用）。
func (h *HealthEngine) snapshotState(name string) string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	if n := h.nodes[name]; n != nil {
		return n.State
	}
	return ""
}

// Candidates 返回候选节点名（candidate 状态，按区域+延迟排序稳定输出）。
func (h *HealthEngine) Candidates() []string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	var out []string
	for _, n := range h.nodes {
		if n.State == stateCandidate {
			out = append(out, n.Name)
		}
	}
	sort.Strings(out)
	return out
}

// RegionEWMA 返回区域组 EWMA（毫秒；无样本区域不在 map 中）。
func (h *HealthEngine) RegionEWMA() map[string]float64 {
	h.mu.RLock()
	defer h.mu.RUnlock()
	out := make(map[string]float64, len(h.byRegion))
	for r, st := range h.byRegion {
		out[r] = math.Round(st.ewma*100) / 100
	}
	return out
}

// NodeDelay 单节点最近延迟（无记录给 -1）。
func (h *HealthEngine) NodeDelay(name string) int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	if n := h.nodes[name]; n != nil {
		return n.LastDelayMs
	}
	return -1
}

// AliveCount 池内存活（candidate）节点数。
func (h *HealthEngine) AliveCount() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	c := 0
	for _, n := range h.nodes {
		if n.State == stateCandidate {
			c++
		}
	}
	return c
}

// TotalCount 池内节点总数（可调度池）。
func (h *HealthEngine) TotalCount() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return len(h.nodes)
}

// RegionOf 节点区域查询。
func (h *HealthEngine) RegionOf(name string) string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	if n := h.nodes[name]; n != nil {
		return n.Region
	}
	return unknownRegion
}

// PoolAlive 池是否存活（至少 1 候选）。
func (h *HealthEngine) PoolAlive() bool { return h.AliveCount() > 0 }

// Snapshots 节点状态快照（/api/proxies 输出，脱敏）。
type NodeSnapshot struct {
	Name        string `json:"name"`
	Type        string `json:"type"`
	Region      string `json:"region"`
	Alive       bool   `json:"alive"`
	DelayMs     int    `json:"delay_ms"`
	ConsecFails int    `json:"consec_fails"`
	State       string `json:"state"`
}

// Snapshot 全量快照（按名称排序）。
func (h *HealthEngine) Snapshot() []NodeSnapshot {
	h.mu.RLock()
	defer h.mu.RUnlock()
	out := make([]NodeSnapshot, 0, len(h.nodes))
	for _, n := range h.nodes {
		out = append(out, NodeSnapshot{
			Name: n.Name, Region: n.Region, Alive: n.State == stateCandidate,
			DelayMs: n.LastDelayMs, ConsecFails: n.ConsecFails, State: n.State,
		})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}