From ccf126de19fc3abb98d46d94d11694d41253965c Mon Sep 17 00:00:00 2001 From: sagitchu Date: Fri, 1 May 2026 06:56:16 +0800 Subject: [PATCH] feat: add best exit selection --- .../internal/http/handler/federation.go | 180 +++++--- .../http/handler/federation_runtime_test.go | 57 +++ go-backend/internal/http/handler/handler.go | 2 + go-backend/internal/http/handler/mutations.go | 161 ++++++- .../internal/http/handler/tunnel_best_exit.go | 432 ++++++++++++++++++ .../http/handler/tunnel_best_exit_test.go | 420 +++++++++++++++++ .../http/handler/tunnel_quality_prober.go | 46 ++ go-gost/x/registry/chain.go | 15 + go-gost/x/registry/chain_test.go | 51 +++ go-gost/x/socket/chain.go | 12 +- go-gost/x/socket/chain_test.go | 69 +++ vite-frontend/src/pages/tunnel.tsx | 3 +- 12 files changed, 1371 insertions(+), 77 deletions(-) create mode 100644 go-backend/internal/http/handler/tunnel_best_exit.go create mode 100644 go-backend/internal/http/handler/tunnel_best_exit_test.go create mode 100644 go-gost/x/registry/chain_test.go create mode 100644 go-gost/x/socket/chain_test.go diff --git a/go-backend/internal/http/handler/federation.go b/go-backend/internal/http/handler/federation.go index 0067bb6..51e1171 100644 --- a/go-backend/internal/http/handler/federation.go +++ b/go-backend/internal/http/handler/federation.go @@ -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 diff --git a/go-backend/internal/http/handler/federation_runtime_test.go b/go-backend/internal/http/handler/federation_runtime_test.go index f821e7a..d544523 100644 --- a/go-backend/internal/http/handler/federation_runtime_test.go +++ b/go-backend/internal/http/handler/federation_runtime_test.go @@ -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 { diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index 571b19e..02ba927 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -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) diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index a52c777..cefa58b 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -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 diff --git a/go-backend/internal/http/handler/tunnel_best_exit.go b/go-backend/internal/http/handler/tunnel_best_exit.go new file mode 100644 index 0000000..6afda57 --- /dev/null +++ b/go-backend/internal/http/handler/tunnel_best_exit.go @@ -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 +} diff --git a/go-backend/internal/http/handler/tunnel_best_exit_test.go b/go-backend/internal/http/handler/tunnel_best_exit_test.go new file mode 100644 index 0000000..e0eb40e --- /dev/null +++ b/go-backend/internal/http/handler/tunnel_best_exit_test.go @@ -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) + } +} diff --git a/go-backend/internal/http/handler/tunnel_quality_prober.go b/go-backend/internal/http/handler/tunnel_quality_prober.go index 4d0bb8a..8a9fba9 100644 --- a/go-backend/internal/http/handler/tunnel_quality_prober.go +++ b/go-backend/internal/http/handler/tunnel_quality_prober.go @@ -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 { diff --git a/go-gost/x/registry/chain.go b/go-gost/x/registry/chain.go index 603d5b5..900dcf2 100644 --- a/go-gost/x/registry/chain.go +++ b/go-gost/x/registry/chain.go @@ -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} diff --git a/go-gost/x/registry/chain_test.go b/go-gost/x/registry/chain_test.go new file mode 100644 index 0000000..5142e87 --- /dev/null +++ b/go-gost/x/registry/chain_test.go @@ -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) + } +} diff --git a/go-gost/x/socket/chain.go b/go-gost/x/socket/chain.go index 4a49a1d..ffc5a76 100644 --- a/go-gost/x/socket/chain.go +++ b/go-gost/x/socket/chain.go @@ -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") } diff --git a/go-gost/x/socket/chain_test.go b/go-gost/x/socket/chain_test.go new file mode 100644 index 0000000..36e6a79 --- /dev/null +++ b/go-gost/x/socket/chain_test.go @@ -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) + } +} diff --git a/vite-frontend/src/pages/tunnel.tsx b/vite-frontend/src/pages/tunnel.tsx index b902143..5f9feee 100644 --- a/vite-frontend/src/pages/tunnel.tsx +++ b/vite-frontend/src/pages/tunnel.tsx @@ -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() { 主备 轮询 随机 + 最优