feat: add best exit selection

This commit is contained in:
sagitchu
2026-05-01 06:56:16 +08:00
parent 4b1be7d3c2
commit ccf126de19
12 changed files with 1371 additions and 77 deletions
+116 -64
View File
@@ -85,6 +85,81 @@ type federationRuntimeReleaseRoleRequest struct {
ResourceKey string `json:"resourceKey"`
}
func federationRuntimeChainName(bindingID string) string {
bindingID = strings.TrimSpace(bindingID)
if bindingID == "" {
return ""
}
return "fed_chain_" + bindingID
}
func buildFederationMiddleChainConfig(chainName string, runtimeID int64, protocol, strategy string, targets []federationRuntimeTarget, interfaceName string) (map[string]interface{}, error) {
chainName = strings.TrimSpace(chainName)
if chainName == "" {
return nil, fmt.Errorf("chain name is required")
}
if len(targets) == 0 {
return nil, fmt.Errorf("targets are required for middle role")
}
protocol = defaultString(protocol, "tls")
nodeItems := make([]map[string]interface{}, 0, len(targets))
for i, target := range targets {
host := strings.TrimSpace(target.Host)
if host == "" || target.Port <= 0 {
return nil, fmt.Errorf("Invalid target")
}
targetProtocol := defaultString(target.Protocol, protocol)
connector := map[string]interface{}{
"type": "relay",
}
if isTCPTunnelProtocol(targetProtocol) {
connector["metadata"] = map[string]interface{}{
"nodelay": true,
"mux.keepaliveInterval": "15s",
"mux.keepaliveTimeout": "45s",
}
}
if isKCPTunnelProtocol(targetProtocol) {
connector["metadata"] = map[string]interface{}{
"connectTimeout": "30s",
}
}
nodeItems = append(nodeItems, map[string]interface{}{
"name": fmt.Sprintf("node_%d", i+1),
"addr": processServerAddress(fmt.Sprintf("%s:%d", host, target.Port)),
"connector": connector,
"dialer": buildTunnelDialerConfig(targetProtocol),
})
}
chainData := map[string]interface{}{
"name": chainName,
"hops": []map[string]interface{}{
{
"name": fmt.Sprintf("hop_%d", runtimeID),
"selector": map[string]interface{}{
"strategy": runtimeTunnelStrategy(strategy),
"maxFails": 1,
"failTimeout": int64(600000000000),
},
"nodes": nodeItems,
},
},
}
if strings.TrimSpace(interfaceName) != "" {
hops := chainData["hops"].([]map[string]interface{})
hops[0]["interface"] = interfaceName
}
return chainData, nil
}
func updateChainPayload(chainName string, chainData map[string]interface{}) map[string]interface{} {
return map[string]interface{}{
"chain": chainName,
"data": chainData,
}
}
type federationRuntimeDiagnoseRequest struct {
IP string `json:"ip"`
Port int `json:"port"`
@@ -145,8 +220,8 @@ type remoteUsageNodeItem struct {
func buildFederationServiceConfig(serviceName, addr, protocol, role, chainName string, targetCount int, interfaceName string) map[string]interface{} {
service := map[string]interface{}{
"name": serviceName,
"addr": addr,
"name": serviceName,
"addr": addr,
"handler": map[string]interface{}{
"type": "relay",
},
@@ -1056,7 +1131,43 @@ func (h *Handler) federationRuntimeApplyRole(w http.ResponseWriter, r *http.Requ
return
}
node, err := h.getNodeRecord(share.NodeID)
if err != nil {
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
}
protocol := defaultString(req.Protocol, runtime.Protocol)
strategy := defaultString(req.Strategy, "round")
chainName := defaultString(runtime.ChainName, federationRuntimeChainName(runtime.BindingID))
if chainName == "" {
chainName = federationRuntimeChainName(fmt.Sprintf("%d", runtime.ID))
}
serviceName := fmt.Sprintf("fed_svc_%d", runtime.ID)
if runtime.Applied == 1 && strings.TrimSpace(runtime.BindingID) != "" {
if req.Role == "middle" && len(req.Targets) > 0 {
chainData, buildErr := buildFederationMiddleChainConfig(chainName, runtime.ID, protocol, strategy, req.Targets, node.InterfaceName)
if buildErr != nil {
response.WriteJSON(w, response.ErrDefault(buildErr.Error()))
return
}
if _, err := h.sendNodeCommand(share.NodeID, "UpdateChains", updateChainPayload(chainName, chainData), false, false); err != nil {
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
}
targetBytes, _ := json.Marshal(req.Targets)
runtime.Role = req.Role
runtime.ChainName = chainName
runtime.Protocol = protocol
runtime.Strategy = strategy
runtime.Target = string(targetBytes)
runtime.Status = 1
runtime.UpdatedTime = time.Now().UnixMilli()
if err := h.repo.UpdatePeerShareRuntime(runtime); err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
}
response.WriteJSON(w, response.OK(map[string]interface{}{
"bindingId": runtime.BindingID,
"allocatedPort": runtime.Port,
@@ -1076,71 +1187,12 @@ func (h *Handler) federationRuntimeApplyRole(w http.ResponseWriter, r *http.Requ
}
}
node, err := h.getNodeRecord(share.NodeID)
if err != nil {
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
}
protocol := defaultString(req.Protocol, runtime.Protocol)
strategy := defaultString(req.Strategy, "round")
chainName := fmt.Sprintf("fed_chain_%d", runtime.ID)
serviceName := fmt.Sprintf("fed_svc_%d", runtime.ID)
if req.Role == "middle" {
if len(req.Targets) == 0 {
response.WriteJSON(w, response.ErrDefault("targets are required for middle role"))
chainData, buildErr := buildFederationMiddleChainConfig(chainName, runtime.ID, protocol, strategy, req.Targets, node.InterfaceName)
if buildErr != nil {
response.WriteJSON(w, response.ErrDefault(buildErr.Error()))
return
}
nodeItems := make([]map[string]interface{}, 0, len(req.Targets))
for i, target := range req.Targets {
host := strings.TrimSpace(target.Host)
if host == "" || target.Port <= 0 {
response.WriteJSON(w, response.ErrDefault("Invalid target"))
return
}
targetProtocol := defaultString(target.Protocol, protocol)
connector := map[string]interface{}{
"type": "relay",
}
if isTCPTunnelProtocol(targetProtocol) {
connector["metadata"] = map[string]interface{}{
"nodelay": true,
"mux.keepaliveInterval": "15s",
"mux.keepaliveTimeout": "45s",
}
}
if isKCPTunnelProtocol(targetProtocol) {
connector["metadata"] = map[string]interface{}{
"connectTimeout": "30s",
}
}
nodeItems = append(nodeItems, map[string]interface{}{
"name": fmt.Sprintf("node_%d", i+1),
"addr": processServerAddress(fmt.Sprintf("%s:%d", host, target.Port)),
"connector": connector,
"dialer": buildTunnelDialerConfig(targetProtocol),
})
}
chainData := map[string]interface{}{
"name": chainName,
"hops": []map[string]interface{}{
{
"name": fmt.Sprintf("hop_%d", runtime.ID),
"selector": map[string]interface{}{
"strategy": strategy,
"maxFails": 1,
"failTimeout": int64(600000000000),
},
"nodes": nodeItems,
},
},
}
if strings.TrimSpace(node.InterfaceName) != "" {
hops := chainData["hops"].([]map[string]interface{})
hops[0]["interface"] = node.InterfaceName
}
if _, err := h.sendNodeCommand(share.NodeID, "AddChains", chainData, true, false); err != nil {
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
@@ -281,6 +281,63 @@ func TestBuildFederationServiceConfig_NonTLSProtocol_NoNodelay(t *testing.T) {
}
}
func TestFederationRuntimeChainNameDerivesFromBindingID(t *testing.T) {
if got := federationRuntimeChainName("12"); got != "fed_chain_12" {
t.Fatalf("expected fed_chain_12, got %q", got)
}
if got := federationRuntimeChainName(" 12 "); got != "fed_chain_12" {
t.Fatalf("expected trimmed fed_chain_12, got %q", got)
}
if got := federationRuntimeChainName(""); got != "" {
t.Fatalf("expected blank binding ID to stay blank, got %q", got)
}
}
func TestBuildFederationMiddleChainConfigUsesExistingChainNameAndBestStrategy(t *testing.T) {
chainData, err := buildFederationMiddleChainConfig("fed_chain_12", 12, "tls", tunnelStrategyBest, []federationRuntimeTarget{
{Host: "10.0.0.31", Port: 30031, Protocol: "tls"},
{Host: "10.0.0.30", Port: 30030, Protocol: "tls"},
}, "")
if err != nil {
t.Fatalf("build chain: %v", err)
}
if chainData["name"] != "fed_chain_12" {
t.Fatalf("expected existing chain name, got %v", chainData["name"])
}
hops := chainData["hops"].([]map[string]interface{})
selector := hops[0]["selector"].(map[string]interface{})
if selector["strategy"] != bestExitRuntimeStrategy {
t.Fatalf("expected best strategy to map to fifo, got %v", selector["strategy"])
}
nodes := hops[0]["nodes"].([]map[string]interface{})
if nodes[0]["addr"] != "10.0.0.31:30031" || nodes[1]["addr"] != "10.0.0.30:30030" {
t.Fatalf("expected target order to be preserved, got %+v", nodes)
}
}
func TestUpdateChainPayloadWrapsChainDataForAgentUpdate(t *testing.T) {
chainData := map[string]interface{}{
"name": "fed_chain_12",
"hops": []map[string]interface{}{},
}
payload := updateChainPayload("fed_chain_12", chainData)
if len(payload) != 2 {
t.Fatalf("expected exact wrapper with 2 keys, got %+v", payload)
}
if payload["chain"] != "fed_chain_12" {
t.Fatalf("expected chain name in wrapper, got %v", payload["chain"])
}
wrappedData, ok := payload["data"].(map[string]interface{})
if !ok {
t.Fatalf("expected wrapped chain data map, got %T", payload["data"])
}
chainData["name"] = "fed_chain_12_updated"
if wrappedData["name"] != "fed_chain_12_updated" {
t.Fatalf("expected wrapper to preserve chainData identity, got %+v", wrappedData)
}
}
func TestFederationRuntimeReservePortRejectsWhenShareFlowExceeded(t *testing.T) {
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db"))
if err != nil {
@@ -52,6 +52,7 @@ type Handler struct {
nodeOnlineRedeploying map[int64]struct{}
qualityProber *tunnelQualityProber
bestExit *bestExitManager
}
const monitorTunnelQualityEnabledConfigKey = "monitor_tunnel_quality_enabled"
@@ -111,6 +112,7 @@ func New(repo *repo.Repository, jwtSecret string) *Handler {
nodeOnlineRedeployAt: make(map[int64]time.Time),
nodeOnlineRedeployQueued: make(map[int64]struct{}),
nodeOnlineRedeploying: make(map[int64]struct{}),
bestExit: newBestExitManager(),
}
h.healthCheck = health.NewChecker(repo, h.wsServer)
h.qualityProber = newTunnelQualityProber(h)
+155 -6
View File
@@ -8,6 +8,7 @@ import (
"encoding/json"
"errors"
"fmt"
"log"
"math/big"
"net"
"net/http"
@@ -3314,6 +3315,8 @@ func (h *Handler) applyFederationRuntime(state *tunnelCreateState, localDomain s
nextTargets := state.OutNodes
if hopIdx+1 < len(state.ChainHops) {
nextTargets = state.ChainHops[hopIdx+1]
} else {
nextTargets = h.orderBestExitTargets(state.TunnelID, chainNode.NodeID, nextTargets)
}
applyTargets := make([]client.RuntimeTarget, 0, len(nextTargets))
for _, target := range nextTargets {
@@ -3343,7 +3346,7 @@ func (h *Handler) applyFederationRuntime(state *tunnelCreateState, localDomain s
ResourceKey: resourceKey,
Role: "middle",
Protocol: defaultString(chainNode.Protocol, "tls"),
Strategy: defaultString(chainNode.Strategy, "round"),
Strategy: runtimeStrategyForTargets(chainNode, nextTargets),
Targets: applyTargets,
}
applyRes, err := fc.ApplyRole(remoteURL, remoteToken, localDomain, applyReq)
@@ -3457,6 +3460,8 @@ func (h *Handler) applyTunnelRuntimeWithMode(state *tunnelCreateState, upsert bo
targets := state.OutNodes
if len(state.ChainHops) > 0 {
targets = state.ChainHops[0]
} else {
targets = h.orderBestExitTargets(state.TunnelID, inNode.NodeID, targets)
}
chainData, err := buildTunnelChainConfig(state.TunnelID, inNode.NodeID, targets, state.Nodes, state.IPPreference)
if err != nil {
@@ -3472,11 +3477,13 @@ func (h *Handler) applyTunnelRuntimeWithMode(state *tunnelCreateState, upsert bo
}
for i, hop := range state.ChainHops {
nextTargets := state.OutNodes
if i+1 < len(state.ChainHops) {
nextTargets = state.ChainHops[i+1]
}
for _, chainNode := range hop {
nextTargets := state.OutNodes
if i+1 < len(state.ChainHops) {
nextTargets = state.ChainHops[i+1]
} else {
nextTargets = h.orderBestExitTargets(state.TunnelID, chainNode.NodeID, nextTargets)
}
node := state.Nodes[chainNode.NodeID]
if node != nil && (node.IsRemote == 1 || node.Status != 1) {
continue
@@ -3530,6 +3537,130 @@ func (h *Handler) applyTunnelChainOnNode(nodeID int64, chainData map[string]inte
return err
}
func (h *Handler) applyBestExitChainOrder(tunnelID, ownerNodeID int64, outNodes []chainNodeRecord, scores []bestExitCandidateScore, ipPreference string) error {
if h == nil {
log.Printf("best_exit: invalid chain update context tunnel=%d owner=%d", tunnelID, ownerNodeID)
return errors.New("invalid best exit chain update context")
}
if tunnelID <= 0 || ownerNodeID <= 0 || len(outNodes) == 0 {
log.Printf("best_exit: invalid chain update input tunnel=%d owner=%d exits=%d", tunnelID, ownerNodeID, len(outNodes))
return fmt.Errorf("invalid best exit chain update input tunnel=%d owner=%d exits=%d", tunnelID, ownerNodeID, len(outNodes))
}
targets := chainRecordsToRuntimeTargets(outNodes)
orderedIDs := make([]int64, 0, len(scores))
for _, score := range scores {
if score.ExitNodeID > 0 {
orderedIDs = append(orderedIDs, score.ExitNodeID)
}
}
targets = orderRuntimeTargetsByNodeID(targets, orderedIDs)
nodes := make(map[int64]*nodeRecord, len(targets)+1)
if owner, err := h.getNodeRecord(ownerNodeID); err == nil && owner != nil {
nodes[ownerNodeID] = owner
}
for _, target := range targets {
if node, err := h.getNodeRecord(target.NodeID); err == nil && node != nil {
nodes[target.NodeID] = node
}
}
owner := nodes[ownerNodeID]
if owner != nil && owner.IsRemote == 1 {
if err := h.applyRemoteBestExitChainOrder(tunnelID, ownerNodeID, owner, targets, nodes, ipPreference); err != nil {
log.Printf("best_exit: update remote federation chain failed tunnel=%d owner=%d err=%v", tunnelID, ownerNodeID, err)
return err
}
log.Printf("best_exit: updated remote federation chain tunnel=%d owner=%d best_exit=%d", tunnelID, ownerNodeID, targets[0].NodeID)
return nil
}
chainData, err := buildTunnelChainConfig(tunnelID, ownerNodeID, targets, nodes, ipPreference)
if err != nil {
log.Printf("best_exit: build chain failed tunnel=%d owner=%d err=%v", tunnelID, ownerNodeID, err)
return err
}
if err := h.applyTunnelChainOnNode(ownerNodeID, chainData, true); err != nil {
log.Printf("best_exit: update chain failed tunnel=%d owner=%d err=%v", tunnelID, ownerNodeID, err)
return err
}
log.Printf("best_exit: updated chain tunnel=%d owner=%d best_exit=%d", tunnelID, ownerNodeID, targets[0].NodeID)
return nil
}
func (h *Handler) applyRemoteBestExitChainOrder(tunnelID, ownerNodeID int64, owner *nodeRecord, targets []tunnelRuntimeNode, nodes map[int64]*nodeRecord, ipPreference string) error {
if h == nil || h.repo == nil || owner == nil {
return errors.New("invalid remote best exit update context")
}
bindings, err := h.repo.ListActiveFederationTunnelBindingsByTunnel(tunnelID)
if err != nil {
return err
}
var binding *repo.FederationTunnelBinding
for i := range bindings {
if bindings[i].NodeID == ownerNodeID && bindings[i].ChainType == 2 && bindings[i].Status == 1 {
binding = &bindings[i]
break
}
}
if binding == nil {
return fmt.Errorf("active federation middle binding not found for tunnel=%d owner=%d", tunnelID, ownerNodeID)
}
remoteURL := strings.TrimSpace(owner.RemoteURL)
if remoteURL == "" {
remoteURL = strings.TrimSpace(binding.RemoteURL)
}
remoteToken := strings.TrimSpace(owner.RemoteToken)
if remoteURL == "" || remoteToken == "" {
return errors.New("远程节点缺少共享配置")
}
applyTargets := make([]client.RuntimeTarget, 0, len(targets))
for _, target := range targets {
targetNode := nodes[target.NodeID]
if targetNode == nil {
return errors.New("节点不存在")
}
host, hostErr := selectTunnelDialHost(owner, targetNode, ipPreference, target.ConnectIP)
if hostErr != nil {
return hostErr
}
if target.Port <= 0 {
return errors.New("节点端口不能为空")
}
applyTargets = append(applyTargets, client.RuntimeTarget{
Host: host,
Port: target.Port,
Protocol: defaultString(target.Protocol, "tls"),
})
}
ownerRuntimeNode := tunnelRuntimeNode{NodeID: ownerNodeID, Protocol: "tls", Strategy: "round", ChainType: 2}
if chainRows, listErr := h.repo.ListChainNodesForTunnel(tunnelID); listErr == nil {
for _, row := range chainRows {
if row.NodeID == ownerNodeID && row.ChainType == 2 {
ownerRuntimeNode = tunnelRuntimeNode{
NodeID: row.NodeID,
Protocol: row.Protocol,
Strategy: row.Strategy,
Inx: int(row.Inx),
ChainType: row.ChainType,
Port: row.Port,
ConnectIP: row.ConnectIP,
}
break
}
}
}
_, err = client.NewFederationClient().ApplyRole(remoteURL, remoteToken, h.federationLocalDomain(), client.RuntimeApplyRoleRequest{
ResourceKey: strings.TrimSpace(binding.ResourceKey),
Role: "middle",
Protocol: defaultString(ownerRuntimeNode.Protocol, "tls"),
Strategy: runtimeStrategyForTargets(ownerRuntimeNode, targets),
Targets: applyTargets,
})
return err
}
func (h *Handler) upsertTunnelChainOnNode(nodeID int64, chainData map[string]interface{}) error {
if h == nil {
return errors.New("invalid tunnel chain context")
@@ -3700,7 +3831,7 @@ func buildTunnelChainConfig(tunnelID int64, fromNodeID int64, targets []tunnelRu
})
}
strategy := defaultString(strings.TrimSpace(targets[0].Strategy), "round")
strategy := runtimeTunnelStrategy(defaultString(strings.TrimSpace(targets[0].Strategy), "round"))
hop := map[string]interface{}{
"name": fmt.Sprintf("hop_%d", tunnelID),
"selector": map[string]interface{}{
@@ -3720,6 +3851,24 @@ func buildTunnelChainConfig(tunnelID int64, fromNodeID int64, targets []tunnelRu
}, nil
}
func (h *Handler) orderBestExitTargets(tunnelID, ownerNodeID int64, targets []tunnelRuntimeNode) []tunnelRuntimeNode {
if len(targets) <= 1 || !isBestTunnelStrategy(targets[0].Strategy) {
return append([]tunnelRuntimeNode(nil), targets...)
}
if h == nil || h.bestExit == nil {
return append([]tunnelRuntimeNode(nil), targets...)
}
return h.bestExit.orderTargets(bestExitOwnerKey{TunnelID: tunnelID, OwnerNodeID: ownerNodeID}, targets)
}
func runtimeStrategyForTargets(owner tunnelRuntimeNode, targets []tunnelRuntimeNode) string {
strategy := defaultString(owner.Strategy, "round")
if len(targets) > 0 {
strategy = defaultString(targets[0].Strategy, strategy)
}
return runtimeTunnelStrategy(strategy)
}
func buildTunnelChainServiceConfig(tunnelID int64, chainNode tunnelRuntimeNode, node *nodeRecord, nextHopCandidateCount int) []map[string]interface{} {
if node == nil {
return nil
@@ -0,0 +1,432 @@
package handler
import (
"errors"
"sort"
"strings"
"sync"
"time"
)
const (
tunnelStrategyBest = "best"
bestExitRuntimeStrategy = "fifo"
bestExitPublicTargetHost = "www.bing.com"
bestExitPublicTargetPort = 443
bestExitLossPenaltyMsPerPercent = 100.0
bestExitConfirmationRounds = 3
bestExitSwitchCooldown = 30 * time.Second
bestExitApplyRetryCooldown = bestExitSwitchCooldown
bestExitMinLatencyAdvantageMs = 20.0
bestExitMinScoreAdvantageRatio = 0.15
)
type bestExitOwnerKey struct {
TunnelID int64
OwnerNodeID int64
}
type bestExitCandidateScore struct {
OwnerNodeID int64
ExitNodeID int64
ExitName string
OwnerToExitLatency float64
ExitToBingLatency float64
OwnerToExitLoss float64
ExitToBingLoss float64
TotalLatency float64
TotalLoss float64
Score float64
Success bool
ErrorMessage string
}
type bestExitSwitchDecision struct {
Switch bool
ExitNodeID int64
Reason string
Scores []bestExitCandidateScore
}
type bestExitProbeFunc func(nodeID int64, ip string, port int, options diagnosisExecOptions) (latency float64, loss float64, err error)
type bestExitProbeResult struct {
latency float64
loss float64
err error
}
type bestExitDecision struct {
AppliedExitNodeID int64
PendingExitNodeID int64
PendingCount int
LastSwitchAt time.Time
LastApplyFailureAt time.Time
LastApplyFailureExitNodeID int64
LastReason string
Scores []bestExitCandidateScore
}
type bestExitManager struct {
mu sync.Mutex
decisions map[bestExitOwnerKey]*bestExitDecision
}
func newBestExitManager() *bestExitManager {
return &bestExitManager{decisions: make(map[bestExitOwnerKey]*bestExitDecision)}
}
func isBestTunnelStrategy(strategy string) bool {
return strings.EqualFold(strings.TrimSpace(strategy), tunnelStrategyBest)
}
func runtimeTunnelStrategy(strategy string) string {
if isBestTunnelStrategy(strategy) {
return bestExitRuntimeStrategy
}
return strategy
}
func scoreBestExitCandidate(ownerNodeID int64, exit chainNodeRecord, ownerLatency, ownerLoss, publicLatency, publicLoss float64) bestExitCandidateScore {
totalLatency := ownerLatency + publicLatency
totalLoss := combineLossPercent(ownerLoss, publicLoss)
return bestExitCandidateScore{
OwnerNodeID: ownerNodeID,
ExitNodeID: exit.NodeID,
ExitName: exit.NodeName,
OwnerToExitLatency: ownerLatency,
ExitToBingLatency: publicLatency,
OwnerToExitLoss: ownerLoss,
ExitToBingLoss: publicLoss,
TotalLatency: totalLatency,
TotalLoss: totalLoss,
Score: totalLatency + totalLoss*bestExitLossPenaltyMsPerPercent,
Success: true,
}
}
func failedBestExitCandidate(ownerNodeID int64, exit chainNodeRecord, message string) bestExitCandidateScore {
return bestExitCandidateScore{
OwnerNodeID: ownerNodeID,
ExitNodeID: exit.NodeID,
ExitName: exit.NodeName,
Success: false,
ErrorMessage: message,
}
}
func combineLossPercent(a, b float64) float64 {
a = clampPercent(a)
b = clampPercent(b)
return (1 - (1-a/100.0)*(1-b/100.0)) * 100.0
}
func clampPercent(v float64) float64 {
if v < 0 {
return 0
}
if v > 100 {
return 100
}
return v
}
func sortBestExitScores(scores []bestExitCandidateScore) {
sort.SliceStable(scores, func(i, j int) bool {
return bestExitScoreLess(scores[i], scores[j])
})
}
func evaluateBestExitOwner(owner chainNodeRecord, exits []chainNodeRecord, nodes map[int64]*nodeRecord, ipPreference string, options diagnosisExecOptions, ping bestExitProbeFunc) []bestExitCandidateScore {
scores := make([]bestExitCandidateScore, 0, len(exits))
if owner.NodeID <= 0 || len(exits) == 0 || ping == nil {
return scores
}
ownerNode := nodes[owner.NodeID]
for _, exit := range exits {
exitNode := nodes[exit.NodeID]
if exitNode == nil {
scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, "exit node unavailable"))
continue
}
targetIP, targetPort, resolveErr := resolveBestExitProbeTarget(ownerNode, exitNode, exit.Port, ipPreference, exit.ConnectIP)
if resolveErr != nil {
scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, resolveErr.Error()))
continue
}
ownerLatency, ownerLoss, ownerErr := ping(owner.NodeID, targetIP, targetPort, options)
if ownerErr != nil {
scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, ownerErr.Error()))
continue
}
publicLatency, publicLoss, publicErr := ping(exit.NodeID, bestExitPublicTargetHost, bestExitPublicTargetPort, options)
if publicErr != nil {
scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, publicErr.Error()))
continue
}
scores = append(scores, scoreBestExitCandidate(owner.NodeID, exit, ownerLatency, ownerLoss, publicLatency, publicLoss))
}
sortBestExitScores(scores)
return scores
}
func resolveBestExitProbeTarget(fromNode, targetNode *nodeRecord, preferredPort int, ipPreference string, connectIP string) (string, int, error) {
if targetNode == nil {
return "", 0, errors.New("目标节点不存在")
}
host, err := selectTunnelDialHost(fromNode, targetNode, ipPreference, connectIP)
if err != nil {
return "", 0, err
}
if strings.TrimSpace(host) == "" {
return "", 0, errors.New("目标节点地址为空")
}
port := preferredPort
if port <= 0 {
port = firstPortFromRange(targetNode.PortRange)
}
if port <= 0 {
port = 443
}
return host, port, nil
}
func newBestExitRoundPinger(base bestExitProbeFunc) bestExitProbeFunc {
cache := make(map[int64]bestExitProbeResult)
return func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) {
if ip == bestExitPublicTargetHost && port == bestExitPublicTargetPort {
if cached, ok := cache[nodeID]; ok {
return cached.latency, cached.loss, cached.err
}
lat, loss, err := base(nodeID, ip, port, options)
cache[nodeID] = bestExitProbeResult{latency: lat, loss: loss, err: err}
return lat, loss, err
}
return base(nodeID, ip, port, options)
}
}
func bestExitChainOwners(inNodes []chainNodeRecord, chainHops [][]chainNodeRecord) []chainNodeRecord {
if len(chainHops) == 0 {
return inNodes
}
return chainHops[len(chainHops)-1]
}
func chainRecordsToRuntimeTargets(rows []chainNodeRecord) []tunnelRuntimeNode {
out := make([]tunnelRuntimeNode, 0, len(rows))
for _, row := range rows {
out = append(out, tunnelRuntimeNode{
NodeID: row.NodeID,
Protocol: row.Protocol,
Strategy: row.Strategy,
Inx: int(row.Inx),
ChainType: row.ChainType,
Port: row.Port,
ConnectIP: row.ConnectIP,
})
}
return out
}
func orderRuntimeTargetsByNodeID(targets []tunnelRuntimeNode, orderedIDs []int64) []tunnelRuntimeNode {
out := append([]tunnelRuntimeNode(nil), targets...)
if len(out) <= 1 || len(orderedIDs) == 0 {
return out
}
positions := make(map[int64]int, len(orderedIDs))
for i, id := range orderedIDs {
if _, ok := positions[id]; !ok {
positions[id] = i
}
}
sort.SliceStable(out, func(i, j int) bool {
pi, iok := positions[out[i].NodeID]
pj, jok := positions[out[j].NodeID]
if iok != jok {
return iok
}
if iok && jok && pi != pj {
return pi < pj
}
return false
})
return out
}
func cloneBestExitScores(scores []bestExitCandidateScore) []bestExitCandidateScore {
return append([]bestExitCandidateScore(nil), scores...)
}
func bestExitDecisionResult(switchNow bool, exitNodeID int64, reason string, scores []bestExitCandidateScore) bestExitSwitchDecision {
return bestExitSwitchDecision{Switch: switchNow, ExitNodeID: exitNodeID, Reason: reason, Scores: cloneBestExitScores(scores)}
}
func bestExitScoreLess(a, b bestExitCandidateScore) bool {
if a.Success != b.Success {
return a.Success
}
if !a.Success && !b.Success {
return a.ExitNodeID < b.ExitNodeID
}
if a.Score != b.Score {
return a.Score < b.Score
}
return a.ExitNodeID < b.ExitNodeID
}
func bestExitHasMinimumAdvantage(candidate, current bestExitCandidateScore) bool {
if !candidate.Success {
return false
}
if !current.Success {
return true
}
improvement := current.Score - candidate.Score
threshold := current.Score * bestExitMinScoreAdvantageRatio
if threshold < bestExitMinLatencyAdvantageMs {
threshold = bestExitMinLatencyAdvantageMs
}
return improvement >= threshold
}
func (m *bestExitManager) setApplied(key bestExitOwnerKey, exitNodeID int64, at time.Time) {
m.mu.Lock()
defer m.mu.Unlock()
d := m.decisionLocked(key)
d.AppliedExitNodeID = exitNodeID
d.PendingExitNodeID = 0
d.PendingCount = 0
d.LastApplyFailureAt = time.Time{}
d.LastApplyFailureExitNodeID = 0
d.LastSwitchAt = at
}
func (m *bestExitManager) recordApplyFailure(key bestExitOwnerKey, exitNodeID int64, at time.Time) {
m.mu.Lock()
defer m.mu.Unlock()
d := m.decisionLocked(key)
d.LastApplyFailureAt = at
d.LastApplyFailureExitNodeID = exitNodeID
d.LastReason = "apply retry cooldown"
}
func (m *bestExitManager) ensureApplied(key bestExitOwnerKey, exitNodeID int64, at time.Time) {
if m == nil || exitNodeID <= 0 {
return
}
m.mu.Lock()
defer m.mu.Unlock()
d := m.decisionLocked(key)
if d.AppliedExitNodeID == 0 {
d.AppliedExitNodeID = exitNodeID
d.LastSwitchAt = at
}
}
func (m *bestExitManager) observeScores(key bestExitOwnerKey, scores []bestExitCandidateScore, now time.Time) bestExitSwitchDecision {
m.mu.Lock()
defer m.mu.Unlock()
ordered := append([]bestExitCandidateScore(nil), scores...)
sortBestExitScores(ordered)
d := m.decisionLocked(key)
d.Scores = cloneBestExitScores(ordered)
if len(ordered) == 0 || !ordered[0].Success {
d.LastReason = "all exits failed"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
candidate := ordered[0]
if d.AppliedExitNodeID == 0 {
d.AppliedExitNodeID = candidate.ExitNodeID
d.LastSwitchAt = now
d.LastReason = "initial best exit"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
if candidate.ExitNodeID == d.AppliedExitNodeID {
d.PendingExitNodeID = 0
d.PendingCount = 0
d.LastApplyFailureAt = time.Time{}
d.LastApplyFailureExitNodeID = 0
d.LastReason = "current exit remains best"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
if candidate.ExitNodeID == d.LastApplyFailureExitNodeID && !d.LastApplyFailureAt.IsZero() && now.Sub(d.LastApplyFailureAt) < bestExitApplyRetryCooldown {
d.LastReason = "apply retry cooldown"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
if now.Sub(d.LastSwitchAt) < bestExitSwitchCooldown {
d.LastReason = "cooldown"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
current := findBestExitScore(ordered, d.AppliedExitNodeID)
if !bestExitHasMinimumAdvantage(candidate, current) {
d.PendingExitNodeID = 0
d.PendingCount = 0
d.LastReason = "insufficient advantage"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
if d.PendingExitNodeID != candidate.ExitNodeID {
d.PendingExitNodeID = candidate.ExitNodeID
d.PendingCount = 1
d.LastReason = "candidate pending confirmation"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
d.PendingCount++
if d.PendingCount < bestExitConfirmationRounds {
d.LastReason = "candidate pending confirmation"
return bestExitDecisionResult(false, 0, d.LastReason, ordered)
}
d.LastReason = "switch confirmed"
return bestExitDecisionResult(true, candidate.ExitNodeID, d.LastReason, ordered)
}
func findBestExitScore(scores []bestExitCandidateScore, exitNodeID int64) bestExitCandidateScore {
for _, score := range scores {
if score.ExitNodeID == exitNodeID {
return score
}
}
return failedBestExitCandidate(0, chainNodeRecord{NodeID: exitNodeID}, "current exit has no successful score")
}
func (m *bestExitManager) decisionLocked(key bestExitOwnerKey) *bestExitDecision {
if d := m.decisions[key]; d != nil {
return d
}
d := &bestExitDecision{}
m.decisions[key] = d
return d
}
func (m *bestExitManager) orderTargets(key bestExitOwnerKey, targets []tunnelRuntimeNode) []tunnelRuntimeNode {
out := append([]tunnelRuntimeNode(nil), targets...)
if m == nil || len(out) <= 1 {
return out
}
m.mu.Lock()
applied := int64(0)
if d := m.decisions[key]; d != nil {
applied = d.AppliedExitNodeID
}
m.mu.Unlock()
if applied <= 0 {
return out
}
sort.SliceStable(out, func(i, j int) bool {
if out[i].NodeID == applied {
return true
}
if out[j].NodeID == applied {
return false
}
return false
})
return out
}
@@ -0,0 +1,420 @@
package handler
import (
"errors"
"testing"
"time"
)
var errBestExitProbeForTest = errors.New("probe failed")
func TestBestExitScoreCombinesLatencyAndLoss(t *testing.T) {
exit := chainNodeRecord{NodeID: 30, NodeName: "exit-a"}
score := scoreBestExitCandidate(10, exit, 25, 2, 80, 3)
if !score.Success {
t.Fatalf("expected successful score")
}
if score.OwnerNodeID != 10 || score.ExitNodeID != 30 {
t.Fatalf("unexpected owner/exit ids: %+v", score)
}
if score.TotalLatency != 105 {
t.Fatalf("expected total latency 105, got %v", score.TotalLatency)
}
if score.TotalLoss < 4.9 || score.TotalLoss > 5.0 {
t.Fatalf("expected combined loss about 4.94, got %v", score.TotalLoss)
}
if score.Score < 599 || score.Score > 600 {
t.Fatalf("expected score about 599, got %v", score.Score)
}
}
func TestBestExitScorePenalizesLoss(t *testing.T) {
stable := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 80, 0, 80, 0)
lowLatencyLossy := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 10, 5, 10, 5)
if !bestExitScoreLess(stable, lowLatencyLossy) {
t.Fatalf("expected stable exit to beat low-latency lossy exit: stable=%+v lossy=%+v", stable, lowLatencyLossy)
}
}
func TestBestExitFailedCandidateSortsLast(t *testing.T) {
failed := failedBestExitCandidate(10, chainNodeRecord{NodeID: 30}, "dial timeout")
good := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 100, 0, 100, 0)
scores := []bestExitCandidateScore{failed, good}
sortBestExitScores(scores)
if scores[0].ExitNodeID != 31 || scores[1].ExitNodeID != 30 {
t.Fatalf("expected good score first and failed score last, got %+v", scores)
}
}
func TestBestExitInitialObservationAppliesWithoutSwitch(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 40, 0, 60, 0)
decision := m.observeScores(key, []bestExitCandidateScore{candidate}, now)
if decision.Switch {
t.Fatalf("initial observation should not return switch: %+v", decision)
}
if m.decisions[key].AppliedExitNodeID != 31 {
t.Fatalf("expected applied exit 31, got %+v", m.decisions[key])
}
}
func TestBestExitDecisionRequiresMinimumAdvantage(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
current := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 100, 0, 100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 90, 0, 90, 0)
m.setApplied(key, 30, now.Add(-time.Minute))
for i := 0; i < bestExitConfirmationRounds+1; i++ {
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(time.Duration(i)*time.Second))
if decision.Switch {
t.Fatalf("candidate below minimum advantage should not switch after repeated observations: %+v", decision)
}
}
if m.decisions[key].AppliedExitNodeID != 30 {
t.Fatalf("expected applied exit to remain 30, got %+v", m.decisions[key])
}
}
func TestBestExitDecisionSwitchesWithMinimumAdvantage(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
current := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 100, 0, 100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 40, 0, 60, 0)
m.setApplied(key, 30, now.Add(-time.Minute))
for i := 0; i < bestExitConfirmationRounds-1; i++ {
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(time.Duration(i)*time.Second))
if decision.Switch {
t.Fatalf("candidate should wait for confirmations before switching: %+v", decision)
}
}
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add((bestExitConfirmationRounds-1)*time.Second))
if !decision.Switch || decision.ExitNodeID != 31 {
t.Fatalf("candidate with enough advantage should switch after confirmations: %+v", decision)
}
}
func TestBestExitConfirmedSwitchDoesNotMarkAppliedUntilSetApplied(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
current := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 100, 0, 100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 40, 0, 60, 0)
m.setApplied(key, 30, now.Add(-time.Minute))
for i := 0; i < bestExitConfirmationRounds-1; i++ {
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(time.Duration(i)*time.Second))
if decision.Switch {
t.Fatalf("candidate should wait for confirmations before switching: %+v", decision)
}
}
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add((bestExitConfirmationRounds-1)*time.Second))
if !decision.Switch || decision.ExitNodeID != 31 {
t.Fatalf("candidate with enough advantage should switch after confirmations: %+v", decision)
}
if m.decisions[key].AppliedExitNodeID != 30 {
t.Fatalf("confirmed switch should not mark applied before runtime update: %+v", m.decisions[key])
}
m.setApplied(key, decision.ExitNodeID, now.Add(time.Second))
if m.decisions[key].AppliedExitNodeID != 31 {
t.Fatalf("setApplied should commit confirmed switch: %+v", m.decisions[key])
}
}
func TestBestExitApplyFailureStartsRetryCooldownWithoutChangingAppliedExit(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
current := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 100, 0, 100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 40, 0, 60, 0)
m.setApplied(key, 30, now.Add(-time.Minute))
for i := 0; i < bestExitConfirmationRounds-1; i++ {
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(time.Duration(i)*time.Second))
if decision.Switch {
t.Fatalf("candidate should wait for confirmations before switching: %+v", decision)
}
}
confirmed := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add((bestExitConfirmationRounds-1)*time.Second))
if !confirmed.Switch || confirmed.ExitNodeID != 31 {
t.Fatalf("expected confirmed switch before apply failure: %+v", confirmed)
}
m.recordApplyFailure(key, confirmed.ExitNodeID, now.Add(bestExitConfirmationRounds*time.Second))
if m.decisions[key].AppliedExitNodeID != 30 {
t.Fatalf("apply failure should leave applied exit unchanged: %+v", m.decisions[key])
}
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add((bestExitConfirmationRounds+1)*time.Second))
if decision.Switch {
t.Fatalf("apply retry cooldown should suppress immediate retry: %+v", decision)
}
if decision.Reason != "apply retry cooldown" {
t.Fatalf("expected apply retry cooldown reason, got %q", decision.Reason)
}
retry := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(bestExitConfirmationRounds*time.Second+bestExitApplyRetryCooldown))
if !retry.Switch || retry.ExitNodeID != 31 {
t.Fatalf("expected retry after apply cooldown: %+v", retry)
}
}
func TestBestExitEnsureAppliedDoesNotOverrideExistingAppliedExit(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
m.ensureApplied(key, 30, now)
if m.decisions[key].AppliedExitNodeID != 30 {
t.Fatalf("expected initial applied exit 30, got %+v", m.decisions[key])
}
if !m.decisions[key].LastSwitchAt.Equal(now) {
t.Fatalf("expected initial applied timestamp, got %+v", m.decisions[key])
}
m.ensureApplied(key, 31, now.Add(time.Minute))
if m.decisions[key].AppliedExitNodeID != 30 {
t.Fatalf("ensureApplied should not override existing applied exit: %+v", m.decisions[key])
}
}
func TestBestExitRoundPingerCachesPublicProbeOnly(t *testing.T) {
publicCalls := 0
ownerCalls := 0
pinger := newBestExitRoundPinger(func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) {
if ip == bestExitPublicTargetHost && port == bestExitPublicTargetPort {
publicCalls++
return float64(nodeID), 0, nil
}
ownerCalls++
return float64(ownerCalls), 0, nil
})
if lat, _, err := pinger(30, bestExitPublicTargetHost, bestExitPublicTargetPort, diagnosisExecOptions{}); err != nil || lat != 30 {
t.Fatalf("unexpected first public ping result lat=%v err=%v", lat, err)
}
if lat, _, err := pinger(30, bestExitPublicTargetHost, bestExitPublicTargetPort, diagnosisExecOptions{}); err != nil || lat != 30 {
t.Fatalf("unexpected cached public ping result lat=%v err=%v", lat, err)
}
if _, _, err := pinger(31, bestExitPublicTargetHost, bestExitPublicTargetPort, diagnosisExecOptions{}); err != nil {
t.Fatalf("unexpected second exit public ping err=%v", err)
}
if publicCalls != 2 {
t.Fatalf("expected public probes cached per exit node, got %d calls", publicCalls)
}
if _, _, err := pinger(10, "10.0.0.30", 30030, diagnosisExecOptions{}); err != nil {
t.Fatalf("unexpected owner ping err=%v", err)
}
if _, _, err := pinger(10, "10.0.0.30", 30030, diagnosisExecOptions{}); err != nil {
t.Fatalf("unexpected repeated owner ping err=%v", err)
}
if ownerCalls != 2 {
t.Fatalf("expected owner-to-exit probes not cached, got %d calls", ownerCalls)
}
}
func TestBestExitDecisionScoresAreDefensiveCopies(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 40, 0, 60, 0)
current := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 100, 0, 100, 0)
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now)
decision.Scores[0].ExitNodeID = 99
if m.decisions[key].Scores[0].ExitNodeID != 31 {
t.Fatalf("decision scores mutation leaked into manager state: %+v", m.decisions[key].Scores)
}
}
func TestBestExitDecisionRequiresConfirmationsAndCooldown(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
now := time.Unix(100, 0)
current := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 100, 0, 100, 0)
candidate := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 31}, 40, 0, 60, 0)
m.setApplied(key, 30, now.Add(-time.Minute))
if decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now); decision.Switch {
t.Fatalf("first observation should not switch: %+v", decision)
}
if decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(time.Second)); decision.Switch {
t.Fatalf("second observation should not switch: %+v", decision)
}
decision := m.observeScores(key, []bestExitCandidateScore{candidate, current}, now.Add(2*time.Second))
if !decision.Switch || decision.ExitNodeID != 31 {
t.Fatalf("third confirmed observation should switch to 31: %+v", decision)
}
betterAgain := scoreBestExitCandidate(10, chainNodeRecord{NodeID: 30}, 20, 0, 20, 0)
if decision := m.observeScores(key, []bestExitCandidateScore{betterAgain, candidate}, now.Add(3*time.Second)); decision.Switch {
t.Fatalf("cooldown should block immediate switch back: %+v", decision)
}
}
func TestBestExitOrderingUsesAppliedDecision(t *testing.T) {
m := newBestExitManager()
key := bestExitOwnerKey{TunnelID: 7, OwnerNodeID: 10}
m.setApplied(key, 31, time.Unix(100, 0))
targets := []tunnelRuntimeNode{
{NodeID: 30, Strategy: tunnelStrategyBest},
{NodeID: 31, Strategy: tunnelStrategyBest},
{NodeID: 32, Strategy: tunnelStrategyBest},
}
ordered := m.orderTargets(key, targets)
if ordered[0].NodeID != 31 || ordered[1].NodeID != 30 || ordered[2].NodeID != 32 {
t.Fatalf("unexpected order: %+v", ordered)
}
if targets[0].NodeID != 30 {
t.Fatalf("orderTargets mutated input: %+v", targets)
}
}
func TestBuildTunnelChainConfigMapsBestStrategyToFIFO(t *testing.T) {
nodes := map[int64]*nodeRecord{
10: {ID: 10, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"},
30: {ID: 30, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"},
31: {ID: 31, ServerIP: "10.0.0.31", ServerIPv4: "10.0.0.31", TCPListenAddr: "[::]"},
}
targets := []tunnelRuntimeNode{
{NodeID: 30, Port: 30030, Protocol: "tls", Strategy: tunnelStrategyBest, ChainType: 3},
{NodeID: 31, Port: 30031, Protocol: "tls", Strategy: tunnelStrategyBest, ChainType: 3},
}
chainData, err := buildTunnelChainConfig(77, 10, targets, nodes, "")
if err != nil {
t.Fatalf("build chain: %v", err)
}
hops := chainData["hops"].([]map[string]interface{})
selector := hops[0]["selector"].(map[string]interface{})
if selector["strategy"] != bestExitRuntimeStrategy {
t.Fatalf("expected best to render as fifo, got %v", selector["strategy"])
}
}
func TestHandlerOrdersBestExitTargetsForOwner(t *testing.T) {
h := &Handler{bestExit: newBestExitManager()}
key := bestExitOwnerKey{TunnelID: 77, OwnerNodeID: 10}
h.bestExit.setApplied(key, 31, time.Unix(100, 0))
targets := []tunnelRuntimeNode{
{NodeID: 30, Port: 30030, Strategy: tunnelStrategyBest},
{NodeID: 31, Port: 30031, Strategy: tunnelStrategyBest},
}
ordered := h.orderBestExitTargets(77, 10, targets)
if ordered[0].NodeID != 31 || ordered[1].NodeID != 30 {
t.Fatalf("unexpected ordered targets: %+v", ordered)
}
}
func TestRuntimeStrategyForTargetsMapsBestTargetStrategyToFIFO(t *testing.T) {
owner := tunnelRuntimeNode{Strategy: "round"}
targets := []tunnelRuntimeNode{{Strategy: tunnelStrategyBest}}
if got := runtimeStrategyForTargets(owner, targets); got != bestExitRuntimeStrategy {
t.Fatalf("expected best target strategy to map to fifo, got %q", got)
}
}
func TestRuntimeStrategyForTargetsPreservesNonBestTargetStrategy(t *testing.T) {
owner := tunnelRuntimeNode{Strategy: tunnelStrategyBest}
targets := []tunnelRuntimeNode{{Strategy: "round"}}
if got := runtimeStrategyForTargets(owner, targets); got != "round" {
t.Fatalf("expected target strategy round to remain unchanged, got %q", got)
}
}
func TestRuntimeStrategyForTargetsMapsBestOwnerStrategyWhenTargetsEmpty(t *testing.T) {
owner := tunnelRuntimeNode{Strategy: tunnelStrategyBest}
if got := runtimeStrategyForTargets(owner, nil); got != bestExitRuntimeStrategy {
t.Fatalf("expected best owner fallback strategy to map to fifo, got %q", got)
}
}
func TestEvaluateBestExitOwnerScoresAllCandidates(t *testing.T) {
owner := chainNodeRecord{NodeID: 10, NodeName: "entry"}
exits := []chainNodeRecord{
{NodeID: 30, NodeName: "exit-a", Port: 30030},
{NodeID: 31, NodeName: "exit-b", Port: 30031},
}
nodes := map[int64]*nodeRecord{
10: {ID: 10, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"},
30: {ID: 30, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"},
31: {ID: 31, ServerIP: "10.0.0.31", ServerIPv4: "10.0.0.31", TCPListenAddr: "[::]"},
}
pinger := func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) {
switch {
case nodeID == 10 && port == 30030:
return 60, 0, nil
case nodeID == 10 && port == 30031:
return 20, 0, nil
case nodeID == 30 && ip == bestExitPublicTargetHost:
return 60, 0, nil
case nodeID == 31 && ip == bestExitPublicTargetHost:
return 20, 0, nil
default:
t.Fatalf("unexpected ping node=%d ip=%s port=%d", nodeID, ip, port)
return 0, 100, nil
}
}
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, pinger)
if len(scores) != 2 {
t.Fatalf("expected two scores, got %+v", scores)
}
if scores[0].ExitNodeID != 31 {
t.Fatalf("expected exit-b first, got %+v", scores)
}
}
func TestEvaluateBestExitOwnerMarksCandidateFailedWhenOwnerToExitFails(t *testing.T) {
owner := chainNodeRecord{NodeID: 10, NodeName: "entry"}
exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-a", Port: 30030}}
nodes := map[int64]*nodeRecord{
10: {ID: 10, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"},
30: {ID: 30, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"},
}
pinger := func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) {
return 0, 100, errBestExitProbeForTest
}
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, pinger)
if len(scores) != 1 || scores[0].Success {
t.Fatalf("expected failed candidate, got %+v", scores)
}
}
func TestEvaluateBestExitOwnerMarksCandidateFailedWhenTargetResolutionFails(t *testing.T) {
owner := chainNodeRecord{NodeID: 10, NodeName: "entry"}
exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-v6", Port: 30030}}
nodes := map[int64]*nodeRecord{
10: {ID: 10, Name: "entry", ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"},
30: {ID: 30, Name: "exit-v6", ServerIP: "2001:db8::30", ServerIPv6: "2001:db8::30", TCPListenAddr: "[::]"},
}
pinger := func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) {
t.Fatalf("ping should not be called when target resolution fails: node=%d ip=%s port=%d", nodeID, ip, port)
return 0, 100, nil
}
scores := evaluateBestExitOwner(owner, exits, nodes, "v4", diagnosisExecOptions{}, pinger)
if len(scores) != 1 || scores[0].Success {
t.Fatalf("expected failed candidate, got %+v", scores)
}
}
@@ -242,6 +242,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
pingTimeoutMS: tunnelQualityPingTimeoutMs,
timeoutMessage: "探测超时",
}
p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options)
switch tunnel.Type {
case 1:
@@ -375,6 +376,51 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
p.storeResult(snap)
}
func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chainNodeRecord, chainHops [][]chainNodeRecord, outNodes []chainNodeRecord, ipPreference string, options diagnosisExecOptions) {
if p == nil || p.handler == nil || p.handler.bestExit == nil || len(outNodes) <= 1 {
return
}
if !isBestTunnelStrategy(outNodes[0].Strategy) {
return
}
owners := bestExitChainOwners(inNodes, chainHops)
if len(owners) == 0 {
return
}
nodeMap := make(map[int64]*nodeRecord, len(owners)+len(outNodes))
for _, owner := range owners {
if node, err := p.handler.getNodeRecord(owner.NodeID); err == nil && node != nil {
nodeMap[owner.NodeID] = node
}
}
for _, exit := range outNodes {
if node, err := p.handler.getNodeRecord(exit.NodeID); err == nil && node != nil {
nodeMap[exit.NodeID] = node
}
}
// This best-exit decision cache is per decision round; the display-oriented
// tunnel quality snapshot may still collect its own first-exit public probe.
roundPinger := newBestExitRoundPinger(p.tcpPingNode)
for _, owner := range owners {
if nodeMap[owner.NodeID] == nil {
continue
}
key := bestExitOwnerKey{TunnelID: tunnelID, OwnerNodeID: owner.NodeID}
p.handler.bestExit.ensureApplied(key, outNodes[0].NodeID, time.Now())
scores := evaluateBestExitOwner(owner, outNodes, nodeMap, ipPreference, options, roundPinger)
decision := p.handler.bestExit.observeScores(key, scores, time.Now())
if decision.Switch {
now := time.Now()
if err := p.handler.applyBestExitChainOrder(tunnelID, owner.NodeID, outNodes, decision.Scores, ipPreference); err != nil {
log.Printf("best_exit: switch apply failed tunnel=%d owner=%d exit=%d err=%v", tunnelID, owner.NodeID, decision.ExitNodeID, err)
p.handler.bestExit.recordApplyFailure(key, decision.ExitNodeID, now)
continue
}
p.handler.bestExit.setApplied(key, decision.ExitNodeID, time.Now())
}
}
}
func (p *tunnelQualityProber) tcpPingNode(nodeID int64, ip string, port int, options diagnosisExecOptions) (latency float64, loss float64, err error) {
h := p.handler
if h == nil {
+15
View File
@@ -12,10 +12,25 @@ type chainRegistry struct {
registry[chain.Chainer]
}
func ReplaceChain(name string, v chain.Chainer) error {
if name == "" {
return nil
}
if r, ok := chainReg.(*chainRegistry); ok {
r.replace(name, v)
return nil
}
return chainReg.Register(name, v)
}
func (r *chainRegistry) Register(name string, v chain.Chainer) error {
return r.registry.Register(name, v)
}
func (r *chainRegistry) replace(name string, v chain.Chainer) {
r.m.Store(name, v)
}
func (r *chainRegistry) Get(name string) chain.Chainer {
if name != "" {
return &chainWrapper{name: name, r: r}
+51
View File
@@ -0,0 +1,51 @@
package registry
import (
"context"
"net"
"testing"
"github.com/go-gost/core/chain"
)
type testChainer struct {
route chain.Route
}
func (c testChainer) Route(context.Context, string, string, ...chain.RouteOption) chain.Route {
return c.route
}
type testRoute struct {
nodes []*chain.Node
}
func (r testRoute) Dial(context.Context, string, string, ...chain.DialOption) (net.Conn, error) {
return nil, nil
}
func (r testRoute) Bind(context.Context, string, string, ...chain.BindOption) (net.Listener, error) {
return nil, nil
}
func (r testRoute) Nodes() []*chain.Node {
return r.nodes
}
func TestReplaceChainOverwritesExistingRegistration(t *testing.T) {
name := "replace_chain_tdd"
ChainRegistry().Unregister(name)
defer ChainRegistry().Unregister(name)
if err := ChainRegistry().Register(name, testChainer{route: testRoute{nodes: []*chain.Node{{Name: "old"}}}}); err != nil {
t.Fatalf("register old chain: %v", err)
}
if err := ReplaceChain(name, testChainer{route: testRoute{nodes: []*chain.Node{{Name: "new"}}}}); err != nil {
t.Fatalf("replace chain: %v", err)
}
route := ChainRegistry().Get(name).Route(context.Background(), "tcp", "example.com:443")
if route == nil || len(route.Nodes()) != 1 || route.Nodes()[0].Name != "new" {
t.Fatalf("expected replacement chain route, got %#v", route)
}
}
+6 -6
View File
@@ -38,21 +38,21 @@ func createChain(req createChainRequest) error {
}
func updateChain(req updateChainRequest) error {
name := strings.TrimSpace(req.Chain)
if registry.ChainRegistry().IsRegistered(name) {
registry.ChainRegistry().Unregister(name)
if name == "" {
name = strings.TrimSpace(req.Data.Name)
}
if name == "" {
return errors.New("chain name is required")
}
req.Data.Name = name
v, err := parser.ParseChain(&req.Data, logger.Default())
if err != nil {
return errors.New("create chain " + name + " failed: " + err.Error())
}
if err := registry.ChainRegistry().Register(name, v); err != nil {
if err := registry.ReplaceChain(name, v); err != nil {
return errors.New("chain " + name + " already exists")
}
+69
View File
@@ -0,0 +1,69 @@
package socket
import (
"testing"
corelogger "github.com/go-gost/core/logger"
"github.com/go-gost/x/config"
_ "github.com/go-gost/x/connector/relay"
_ "github.com/go-gost/x/dialer/tcp"
xlogger "github.com/go-gost/x/logger"
"github.com/go-gost/x/registry"
)
func TestUpdateChainParseFailureKeepsExistingChainRegistered(t *testing.T) {
corelogger.SetDefault(xlogger.Nop())
name := "chain_update_parse_failure_tdd"
originalConfig := config.Global()
defer config.Set(originalConfig)
registry.ChainRegistry().Unregister(name)
defer registry.ChainRegistry().Unregister(name)
config.Set(&config.Config{})
valid := config.ChainConfig{
Name: name,
Hops: []*config.HopConfig{{
Name: "hop-valid",
Nodes: []*config.NodeConfig{{
Name: "node-valid",
Addr: "127.0.0.1:443",
Connector: &config.ConnectorConfig{Type: "relay"},
Dialer: &config.DialerConfig{Type: "tcp"},
}},
}},
}
if err := createChain(createChainRequest{Data: valid}); err != nil {
t.Fatalf("create valid chain: %v", err)
}
before := registry.ChainRegistry().Get(name)
if before == nil || !registry.ChainRegistry().IsRegistered(name) {
t.Fatalf("expected chain registered before update")
}
invalid := config.ChainConfig{
Hops: []*config.HopConfig{{
Name: "hop-invalid",
Nodes: []*config.NodeConfig{{
Name: "node-invalid",
Addr: "127.0.0.1:443",
Connector: &config.ConnectorConfig{Type: "connector-does-not-exist"},
Dialer: &config.DialerConfig{Type: "tcp"},
}},
}},
}
err := updateChain(updateChainRequest{Chain: name, Data: invalid})
if err == nil {
t.Fatalf("expected invalid chain update to fail")
}
if !registry.ChainRegistry().IsRegistered(name) {
t.Fatalf("expected old chain to remain registered after failed update")
}
cfg := config.Global()
if len(cfg.Chains) != 1 || cfg.Chains[0] == nil || cfg.Chains[0].Name != name {
t.Fatalf("expected original chain config to remain, got %#v", cfg.Chains)
}
if got := cfg.Chains[0].Hops[0].Name; got != "hop-valid" {
t.Fatalf("expected original chain config to remain, got hop %q", got)
}
}
+2 -1
View File
@@ -91,7 +91,7 @@ import {
interface ChainTunnel {
nodeId: number;
protocol?: string; // 'tls' | 'wss' | 'tcp' | 'mtls' | 'mwss' | 'mtcp' | 'kcp' - 转发链协议
strategy?: string; // 'fifo' | 'round' | 'rand' - 仅转发链需要
strategy?: string; // 'fifo' | 'round' | 'rand' | 'best' - 仅转发链/多出口需要
chainType?: number; // 1: 入口, 2: 转发链, 3: 出口
inx?: number; // 转发链序号
connectIp?: string; // 连接IP(多IP节点指定连接地址)
@@ -2901,6 +2901,7 @@ export default function TunnelPage() {
<SelectItem key="fifo">主备</SelectItem>
<SelectItem key="round">轮询</SelectItem>
<SelectItem key="rand">随机</SelectItem>
<SelectItem key="best">最优</SelectItem>
</Select>
</div>