From cbe2fc492e7246b05955a35d7b9bd9785b760a4b Mon Sep 17 00:00:00 2001
From: sagit <36596628+Sagit-chu@users.noreply.github.com>
Date: Mon, 3 Aug 2026 10:15:22 +0800
Subject: [PATCH] feat(monitor): show backup tunnel latencies (#535)
Closes #508
---
.../http/handler/tunnel_quality_prober.go | 246 ++++++++++++++++--
.../handler/tunnel_quality_prober_test.go | 52 ++++
.../store/repo/repository_monitoring_test.go | 28 ++
.../store/repo/repository_tunnel_quality.go | 3 +-
vite-frontend/src/api/types.ts | 14 +
.../src/pages/node/tunnel-monitor-view.tsx | 221 ++++++++++++----
6 files changed, 488 insertions(+), 76 deletions(-)
diff --git a/go-backend/internal/http/handler/tunnel_quality_prober.go b/go-backend/internal/http/handler/tunnel_quality_prober.go
index 4c18082..faae91e 100644
--- a/go-backend/internal/http/handler/tunnel_quality_prober.go
+++ b/go-backend/internal/http/handler/tunnel_quality_prober.go
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"log"
"sync"
"sync/atomic"
@@ -31,6 +32,26 @@ type TunnelQualityHop struct {
TargetPort int `json:"targetPort,omitempty"`
}
+type TunnelQualityCandidateHop struct {
+ TunnelQualityHop
+ FromRole string `json:"fromRole"`
+ ToRole string `json:"toRole"`
+ HopIndex int `json:"hopIndex"`
+ Selected bool `json:"selected"`
+ ErrorMessage string `json:"errorMessage,omitempty"`
+}
+
+type tunnelQualityChainDetails struct {
+ PrimaryPath []TunnelQualityHop `json:"primaryPath,omitempty"`
+ CandidateHops []TunnelQualityCandidateHop `json:"candidateHops,omitempty"`
+}
+
+type tunnelQualityCandidateGroup struct {
+ role string
+ roleIndex int
+ nodes []chainNodeRecord
+}
+
// tunnelQualitySnapshot is the in-memory latest probe result for a tunnel.
type tunnelQualitySnapshot struct {
TunnelID int64 `json:"tunnelId"`
@@ -283,16 +304,25 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
pingCount: 1,
timeoutMessage: "探测超时",
}
- p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options, probeTarget)
+ roundPinger := newBestExitRoundPinger(p.pingNode)
+ p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options, probeTarget, roundPinger)
entry, _, entryOnline := p.firstOnlineChainNode(inNodes)
exit, _, exitOnline := p.firstOnlineChainNode(outNodes)
+ selectedNodeIDs := make(map[string]int64, 2+len(midNodesGrouped))
+ if entryOnline {
+ selectedNodeIDs[tunnelQualityGroupKey("entry", 0)] = entry.NodeID
+ }
+ if exitOnline {
+ selectedNodeIDs[tunnelQualityGroupKey("exit", 0)] = exit.NodeID
+ }
+ var primaryHops []TunnelQualityHop
switch tunnel.Type {
case 1:
// Port forwarding: entry → public probe target only.
if entryOnline {
- lat, loss, err := p.pingNode(entry.NodeID, probeTarget.Host, probeTarget.Port, options)
+ lat, loss, err := roundPinger(entry.NodeID, probeTarget.Host, probeTarget.Port, options)
if err == nil {
snap.ExitToBingLatency = lat
snap.ExitToBingLoss = loss
@@ -318,13 +348,12 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
snap.EntryToExitLatency = -1
snap.EntryToExitLoss = 100
} else {
- var hops []TunnelQualityHop
var totalLat float64
remainingSuccessProb := 1.0
nodesInPath := make([]chainNodeRecord, 0, 2+len(midNodesGrouped))
nodesInPath = append(nodesInPath, entry)
- for _, midGroup := range midNodesGrouped {
+ for midIndex, midGroup := range midNodesGrouped {
mid, _, online := p.firstOnlineChainNode(midGroup)
if !online {
probeOK = false
@@ -332,6 +361,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
break
}
nodesInPath = append(nodesInPath, mid)
+ selectedNodeIDs[tunnelQualityGroupKey("middle", midIndex)] = mid.NodeID
}
if probeOK {
nodesInPath = append(nodesInPath, exit)
@@ -354,7 +384,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
probeOK = false
hop.Latency = -1
hop.Loss = 100
- hops = append(hops, hop)
+ primaryHops = append(primaryHops, hop)
break
}
@@ -365,25 +395,25 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
probeOK = false
hop.Latency = -1
hop.Loss = 100
- hops = append(hops, hop)
+ primaryHops = append(primaryHops, hop)
break
}
hop.TargetIP = targetIP
hop.TargetPort = targetPort
- lat, loss, err := p.pingNode(source.NodeID, targetIP, targetPort, options)
+ lat, loss, err := roundPinger(source.NodeID, targetIP, targetPort, options)
if err == nil {
hop.Latency = lat
hop.Loss = loss
totalLat += lat
remainingSuccessProb *= (1.0 - loss/100.0)
- hops = append(hops, hop)
+ primaryHops = append(primaryHops, hop)
} else {
probeOK = false
hop.Latency = -1
hop.Loss = 100
- hops = append(hops, hop)
+ primaryHops = append(primaryHops, hop)
if snap.ErrorMessage == "" {
snap.ErrorMessage = err.Error()
}
@@ -398,17 +428,11 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
snap.EntryToExitLatency = -1
snap.EntryToExitLoss = 100
}
-
- if len(hops) > 0 {
- if b, err := json.Marshal(hops); err == nil {
- snap.ChainDetails = string(b)
- }
- }
}
// Exit → Bing
if exitOnline {
- lat, loss, err := p.pingNode(exit.NodeID, probeTarget.Host, probeTarget.Port, options)
+ lat, loss, err := roundPinger(exit.NodeID, probeTarget.Host, probeTarget.Port, options)
if err == nil {
snap.ExitToBingLatency = lat
snap.ExitToBingLoss = loss
@@ -424,7 +448,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
default:
// Unknown type: entry → public probe target.
if entryOnline {
- lat, loss, err := p.pingNode(entry.NodeID, probeTarget.Host, probeTarget.Port, options)
+ lat, loss, err := roundPinger(entry.NodeID, probeTarget.Host, probeTarget.Port, options)
if err == nil {
snap.ExitToBingLatency = lat
snap.ExitToBingLoss = loss
@@ -437,9 +461,192 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
}
}
+ candidateHops := p.probeTunnelCandidateHops(
+ tunnel.Type,
+ inNodes,
+ midNodesGrouped,
+ outNodes,
+ selectedNodeIDs,
+ ipPreference,
+ options,
+ probeTarget,
+ roundPinger,
+ )
+ if len(primaryHops) > 0 || len(candidateHops) > 0 {
+ details := tunnelQualityChainDetails{
+ PrimaryPath: primaryHops,
+ CandidateHops: candidateHops,
+ }
+ if b, err := json.Marshal(details); err == nil {
+ snap.ChainDetails = string(b)
+ }
+ }
+
p.storeResult(snap)
}
+func tunnelQualityGroupKey(role string, index int) string {
+ return fmt.Sprintf("%s:%d", role, index)
+}
+
+func (p *tunnelQualityProber) probeTunnelCandidateHops(
+ tunnelType int,
+ inNodes []chainNodeRecord,
+ chainHops [][]chainNodeRecord,
+ outNodes []chainNodeRecord,
+ selectedNodeIDs map[string]int64,
+ ipPreference string,
+ options diagnosisExecOptions,
+ probeTarget tunnelProbeTarget,
+ ping bestExitProbeFunc,
+) []TunnelQualityCandidateHop {
+ if p == nil || p.handler == nil || ping == nil {
+ return nil
+ }
+
+ if tunnelType != 2 {
+ return p.probePublicTargetCandidates("entry", 0, inNodes, selectedNodeIDs, options, probeTarget, ping)
+ }
+
+ groups := make([]tunnelQualityCandidateGroup, 0, 2+len(chainHops))
+ groups = append(groups, tunnelQualityCandidateGroup{role: "entry", roleIndex: 0, nodes: inNodes})
+ for i, hop := range chainHops {
+ groups = append(groups, tunnelQualityCandidateGroup{role: "middle", roleIndex: i, nodes: hop})
+ }
+ groups = append(groups, tunnelQualityCandidateGroup{role: "exit", roleIndex: 0, nodes: outNodes})
+
+ var items []TunnelQualityCandidateHop
+ for i := 0; i < len(groups)-1; i++ {
+ items = append(items, p.probeCandidateGroupLinks(
+ groups[i],
+ groups[i+1],
+ i,
+ selectedNodeIDs,
+ ipPreference,
+ options,
+ ping,
+ )...)
+ }
+ items = append(items, p.probePublicTargetCandidates(
+ "exit",
+ 0,
+ outNodes,
+ selectedNodeIDs,
+ options,
+ probeTarget,
+ ping,
+ )...)
+ return items
+}
+
+func (p *tunnelQualityProber) probeCandidateGroupLinks(
+ fromGroup tunnelQualityCandidateGroup,
+ toGroup tunnelQualityCandidateGroup,
+ hopIndex int,
+ selectedNodeIDs map[string]int64,
+ ipPreference string,
+ options diagnosisExecOptions,
+ ping bestExitProbeFunc,
+) []TunnelQualityCandidateHop {
+ items := make([]TunnelQualityCandidateHop, 0, len(fromGroup.nodes)*len(toGroup.nodes))
+ for _, source := range fromGroup.nodes {
+ for _, target := range toGroup.nodes {
+ item := TunnelQualityCandidateHop{
+ TunnelQualityHop: TunnelQualityHop{
+ FromNodeID: source.NodeID,
+ FromNodeName: source.NodeName,
+ ToNodeID: target.NodeID,
+ ToNodeName: target.NodeName,
+ Latency: -1,
+ Loss: 100,
+ },
+ FromRole: fromGroup.role,
+ ToRole: toGroup.role,
+ HopIndex: hopIndex,
+ Selected: selectedNodeIDs[tunnelQualityGroupKey(fromGroup.role, fromGroup.roleIndex)] == source.NodeID &&
+ selectedNodeIDs[tunnelQualityGroupKey(toGroup.role, toGroup.roleIndex)] == target.NodeID,
+ }
+
+ sourceNode, sourceErr := p.handler.getNodeRecord(source.NodeID)
+ if sourceErr != nil || !isTunnelProbeNodeOnline(sourceNode) {
+ item.ErrorMessage = "来源节点不在线"
+ items = append(items, item)
+ continue
+ }
+ targetNode, targetErr := p.handler.getNodeRecord(target.NodeID)
+ if targetErr != nil || !isTunnelProbeNodeOnline(targetNode) {
+ item.ErrorMessage = "目标节点不在线"
+ items = append(items, item)
+ continue
+ }
+
+ targetIP, targetPort, resolveErr := resolveChainProbeTarget(sourceNode, targetNode, target.Port, ipPreference, target.ConnectIP)
+ if resolveErr != nil {
+ item.ErrorMessage = resolveErr.Error()
+ items = append(items, item)
+ continue
+ }
+ item.TargetIP = targetIP
+ item.TargetPort = targetPort
+ latency, loss, probeErr := ping(source.NodeID, targetIP, targetPort, options)
+ if probeErr != nil {
+ item.ErrorMessage = probeErr.Error()
+ items = append(items, item)
+ continue
+ }
+ item.Latency = latency
+ item.Loss = loss
+ items = append(items, item)
+ }
+ }
+ return items
+}
+
+func (p *tunnelQualityProber) probePublicTargetCandidates(
+ fromRole string,
+ fromIndex int,
+ nodes []chainNodeRecord,
+ selectedNodeIDs map[string]int64,
+ options diagnosisExecOptions,
+ probeTarget tunnelProbeTarget,
+ ping bestExitProbeFunc,
+) []TunnelQualityCandidateHop {
+ items := make([]TunnelQualityCandidateHop, 0, len(nodes))
+ for _, source := range nodes {
+ item := TunnelQualityCandidateHop{
+ TunnelQualityHop: TunnelQualityHop{
+ FromNodeID: source.NodeID,
+ FromNodeName: source.NodeName,
+ ToNodeName: formatTunnelProbeTarget(probeTarget),
+ Latency: -1,
+ Loss: 100,
+ TargetIP: probeTarget.Host,
+ TargetPort: probeTarget.Port,
+ },
+ FromRole: fromRole,
+ ToRole: "target",
+ HopIndex: fromIndex,
+ Selected: selectedNodeIDs[tunnelQualityGroupKey(fromRole, fromIndex)] == source.NodeID,
+ }
+ sourceNode, sourceErr := p.handler.getNodeRecord(source.NodeID)
+ if sourceErr != nil || !isTunnelProbeNodeOnline(sourceNode) {
+ item.ErrorMessage = "来源节点不在线"
+ items = append(items, item)
+ continue
+ }
+ latency, loss, probeErr := ping(source.NodeID, probeTarget.Host, probeTarget.Port, options)
+ if probeErr != nil {
+ item.ErrorMessage = probeErr.Error()
+ items = append(items, item)
+ continue
+ }
+ item.Latency = latency
+ item.Loss = loss
+ items = append(items, item)
+ }
+ return items
+}
+
func isTunnelProbeNodeOnline(node *nodeRecord) bool {
return node != nil && (node.IsRemote == 1 || node.Status == 1)
}
@@ -457,7 +664,7 @@ func (p *tunnelQualityProber) firstOnlineChainNode(nodes []chainNodeRecord) (cha
return chainNodeRecord{}, nil, false
}
-func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chainNodeRecord, chainHops [][]chainNodeRecord, outNodes []chainNodeRecord, ipPreference string, options diagnosisExecOptions, probeTarget tunnelProbeTarget) {
+func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chainNodeRecord, chainHops [][]chainNodeRecord, outNodes []chainNodeRecord, ipPreference string, options diagnosisExecOptions, probeTarget tunnelProbeTarget, roundPinger bestExitProbeFunc) {
if p == nil || p.handler == nil || p.handler.bestExit == nil || len(outNodes) <= 1 {
return
}
@@ -479,9 +686,6 @@ func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chai
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.pingNode)
for _, owner := range owners {
if nodeMap[owner.NodeID] == nil {
continue
diff --git a/go-backend/internal/http/handler/tunnel_quality_prober_test.go b/go-backend/internal/http/handler/tunnel_quality_prober_test.go
index b66c20b..3953522 100644
--- a/go-backend/internal/http/handler/tunnel_quality_prober_test.go
+++ b/go-backend/internal/http/handler/tunnel_quality_prober_test.go
@@ -1,6 +1,7 @@
package handler
import (
+ "encoding/json"
"fmt"
"slices"
"testing"
@@ -106,6 +107,57 @@ func TestTunnelQualityProberUsesOnlineBackupExit(t *testing.T) {
}
}
+func TestTunnelQualityProberReportsAllExitCandidateLatencies(t *testing.T) {
+ h := setupProbeTargetTunnelHandler(t)
+ seedQualityForwardTunnel(t, h, 83, []int{1, 1})
+
+ p := newTunnelQualityProber(h)
+ p.probeNode = func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) {
+ switch fmt.Sprintf("%d|%s|%d", nodeID, ip, port) {
+ case "10|10.0.0.30|30030":
+ return 20, 0, nil
+ case "10|10.0.0.31|30031":
+ return 35, 0, nil
+ case "30|www.bing.com|443":
+ return 50, 0, nil
+ case "31|www.bing.com|443":
+ return 65, 0, nil
+ default:
+ return 0, 100, fmt.Errorf("unexpected probe node=%d target=%s:%d", nodeID, ip, port)
+ }
+ }
+ p.probeTunnel(83)
+
+ snaps := p.GetAll()
+ if len(snaps) != 1 {
+ t.Fatalf("expected one quality snapshot, got %+v", snaps)
+ }
+ if snaps[0].EntryToExitLatency != 20 || snaps[0].ExitToBingLatency != 50 {
+ t.Fatalf("expected primary path metrics to remain unchanged, got %+v", snaps[0])
+ }
+
+ var details tunnelQualityChainDetails
+ if err := json.Unmarshal([]byte(snaps[0].ChainDetails), &details); err != nil {
+ t.Fatalf("decode chain details: %v", err)
+ }
+ assertCandidateHop := func(fromID, toID int64, latency float64, selected bool) {
+ t.Helper()
+ for _, hop := range details.CandidateHops {
+ if hop.FromNodeID == fromID && hop.ToNodeID == toID {
+ if hop.Latency != latency || hop.Selected != selected || hop.ErrorMessage != "" {
+ t.Fatalf("unexpected candidate hop: %+v", hop)
+ }
+ return
+ }
+ }
+ t.Fatalf("candidate hop %d -> %d not found in %+v", fromID, toID, details.CandidateHops)
+ }
+ assertCandidateHop(10, 30, 20, true)
+ assertCandidateHop(10, 31, 35, false)
+ assertCandidateHop(30, 0, 50, true)
+ assertCandidateHop(31, 0, 65, false)
+}
+
func TestTunnelQualityProberStoresProbeTargetWhenChainIncomplete(t *testing.T) {
h := setupProbeTargetTunnelHandler(t)
seedProbeTargetTunnel(t, h, 78, "quality-target-incomplete", "speed.example.com", 8443)
diff --git a/go-backend/internal/store/repo/repository_monitoring_test.go b/go-backend/internal/store/repo/repository_monitoring_test.go
index 2d624ff..c9c68a2 100644
--- a/go-backend/internal/store/repo/repository_monitoring_test.go
+++ b/go-backend/internal/store/repo/repository_monitoring_test.go
@@ -132,3 +132,31 @@ func TestUpsertTunnelMetricBucketsIsSafeUnderConcurrency(t *testing.T) {
t.Fatalf("expected bytesOut %d, got %d", wantOut, rows[0].BytesOut)
}
}
+
+func TestGetLatestTunnelQualitiesIncludesChainDetails(t *testing.T) {
+ r, err := Open(":memory:")
+ if err != nil {
+ t.Fatalf("open repo: %v", err)
+ }
+ defer r.Close()
+
+ if err := r.InsertTunnelQuality(&model.TunnelQuality{
+ TunnelID: 7,
+ Timestamp: time.Now().UnixMilli(),
+ Success: 1,
+ ChainDetails: `{"primaryPath":[],"candidateHops":[{"fromNodeId":10,"toNodeId":31}]}`,
+ }); err != nil {
+ t.Fatalf("insert tunnel quality: %v", err)
+ }
+
+ items, err := r.GetLatestTunnelQualities()
+ if err != nil {
+ t.Fatalf("get latest tunnel qualities: %v", err)
+ }
+ if len(items) != 1 {
+ t.Fatalf("expected one latest tunnel quality, got %+v", items)
+ }
+ if items[0].ChainDetails == "" {
+ t.Fatalf("expected chain details in latest quality row, got %+v", items[0])
+ }
+}
diff --git a/go-backend/internal/store/repo/repository_tunnel_quality.go b/go-backend/internal/store/repo/repository_tunnel_quality.go
index de5f4bb..c5cbe37 100644
--- a/go-backend/internal/store/repo/repository_tunnel_quality.go
+++ b/go-backend/internal/store/repo/repository_tunnel_quality.go
@@ -44,7 +44,8 @@ func (r *Repository) GetLatestTunnelQualities() ([]model.TunnelQuality, error) {
// Use window function (works on modern SQLite 3.25+ and PostgreSQL).
q := `
SELECT id, tunnel_id, entry_to_exit_latency, exit_to_bing_latency,
- entry_to_exit_loss, exit_to_bing_loss, success, error_message, timestamp
+ entry_to_exit_loss, exit_to_bing_loss, success, error_message, timestamp,
+ chain_details
FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY tunnel_id ORDER BY timestamp DESC, id DESC) AS rn
FROM tunnel_quality
diff --git a/vite-frontend/src/api/types.ts b/vite-frontend/src/api/types.ts
index 37bd990..5cdc3b9 100644
--- a/vite-frontend/src/api/types.ts
+++ b/vite-frontend/src/api/types.ts
@@ -595,6 +595,20 @@ export interface TunnelQualityHopApiItem {
targetPort?: number;
}
+export interface TunnelQualityCandidateHopApiItem
+ extends TunnelQualityHopApiItem {
+ fromRole: "entry" | "middle" | "exit";
+ toRole: "middle" | "exit" | "target";
+ hopIndex: number;
+ selected: boolean;
+ errorMessage?: string;
+}
+
+export interface TunnelQualityChainDetailsApiItem {
+ primaryPath?: TunnelQualityHopApiItem[];
+ candidateHops?: TunnelQualityCandidateHopApiItem[];
+}
+
export interface TunnelQualityApiItem {
tunnelId: number;
entryToExitLatency: number;
diff --git a/vite-frontend/src/pages/node/tunnel-monitor-view.tsx b/vite-frontend/src/pages/node/tunnel-monitor-view.tsx
index f54d2ce..3ad6f22 100644
--- a/vite-frontend/src/pages/node/tunnel-monitor-view.tsx
+++ b/vite-frontend/src/pages/node/tunnel-monitor-view.tsx
@@ -2,6 +2,8 @@ import type {
MonitorTunnelApiItem,
TunnelMetricApiItem,
TunnelQualityApiItem,
+ TunnelQualityCandidateHopApiItem,
+ TunnelQualityChainDetailsApiItem,
TunnelQualityHopApiItem,
} from "@/api/types";
@@ -468,18 +470,59 @@ const TrafficChartCard = React.memo(function TrafficChartCard({
);
});
-function ForwardingChainTopology({ hopsStr }: { hopsStr?: string }) {
- if (!hopsStr) return null;
-
- let hops: TunnelQualityHopApiItem[] = [];
-
+const parseTunnelQualityChainDetails = (
+ raw?: string,
+): TunnelQualityChainDetailsApiItem => {
+ if (!raw) return {};
try {
- hops = JSON.parse(hopsStr);
+ const parsed: unknown = JSON.parse(raw);
+
+ // Backward-compatible with historical rows that stored the primary path
+ // directly as a JSON array.
+ if (Array.isArray(parsed)) {
+ return { primaryPath: parsed as TunnelQualityHopApiItem[] };
+ }
+ if (parsed && typeof parsed === "object") {
+ return parsed as TunnelQualityChainDetailsApiItem;
+ }
} catch {
- return null;
+ return {};
}
- if (!Array.isArray(hops) || hops.length === 0) return null;
+ return {};
+};
+
+const candidateRoleLabel = (role: string) => {
+ switch (role) {
+ case "entry":
+ return "入口";
+ case "middle":
+ return "中转";
+ case "exit":
+ return "出口";
+ case "target":
+ return "测试目标";
+ default:
+ return role;
+ }
+};
+
+const ForwardingChainTopology = React.memo(function ForwardingChainTopology({
+ hopsStr,
+}: {
+ hopsStr?: string;
+}) {
+ const details = useMemo(
+ () => parseTunnelQualityChainDetails(hopsStr),
+ [hopsStr],
+ );
+ const primaryHops = details.primaryPath ?? [];
+ const alternativeHops = useMemo(
+ () => (details.candidateHops ?? []).filter((hop) => !hop.selected),
+ [details.candidateHops],
+ );
+
+ if (primaryHops.length === 0 && alternativeHops.length === 0) return null;
return (