fix: resolve federation forward card showing zero flow (#212)

## Summary

Fixed federation port-forward tunnels not displaying traffic on forward
cards.

**Root Cause:**
- Frontend's mergeFederationShareFlow only parsed shareId from tunnel
name (Share-{id}-Port-{port} format)
- Federation tunnels with custom names couldn't be resolved, causing
traffic display to show 0

**Fix:**
- Added fallback to resolve shareId via remote-usage bindings[].tunnelId
- Multiple resolution candidates are merged with max(currentFlow)
selection
- Maintains directFlow precedence over federation share flow

**Tests:**
- Added contract test for custom tunnel name with binding-based
resolution
- All existing tests pass
- Frontend builds successfully

Fixes: federation port-forward showing total flow 0
This commit is contained in:
sagit
2026-02-25 21:24:50 +08:00
committed by GitHub
2 changed files with 295 additions and 13 deletions
@@ -386,6 +386,221 @@ func TestFederationForwardCardFlowLinkageContractSplitShareFlowAcrossMultipleFor
} }
} }
func TestFederationForwardCardFlowLinkageContractResolvesShareByTunnelBindingWhenTunnelNameIsCustom(t *testing.T) {
secret := "federation-forward-binding-flow-contract-jwt"
router, r := setupContractRouter(t, secret)
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
if err != nil {
t.Fatalf("generate admin token: %v", err)
}
now := time.Now().UnixMilli()
remoteShareID := int64(901)
remoteShareFlow := int64(5000)
if err := r.DB().Exec(`
INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx, is_remote, remote_url, remote_token, remote_config)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
"flow-binding-remote-node", "flow-binding-remote-secret", "10.31.41.51", "10.31.41.51", "", "33000-33020", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 1, "", "", fmt.Sprintf(`{"shareId":%d,"maxBandwidth":0,"currentFlow":%d,"portRangeStart":33000,"portRangeEnd":33020}`, remoteShareID, remoteShareFlow),
).Error; err != nil {
t.Fatalf("insert remote node: %v", err)
}
remoteNodeID := mustLastInsertID(t, r, "flow-binding-remote-node")
if err := r.DB().Exec(`
INSERT INTO tunnel(name, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)
`, "federation-port-forward-custom-name", 1, "tcp", 1, now, now, 1, "", 0).Error; err != nil {
t.Fatalf("insert custom tunnel: %v", err)
}
tunnelID := mustLastInsertID(t, r, "flow-binding-custom-tunnel")
if err := r.DB().Exec(`
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, 1, "admin_user", "flow-binding-forward", tunnelID, "1.1.1.1:443", "fifo", 0, 0, now, now, 1, 0).Error; err != nil {
t.Fatalf("insert forward: %v", err)
}
forwardID := mustLastInsertID(t, r, "flow-binding-forward")
if err := r.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, remoteNodeID, 33001).Error; err != nil {
t.Fatalf("insert forward_port: %v", err)
}
forwardOut := requestContractEnvelope(t, router, adminToken, "/api/v1/forward/list", nil)
if forwardOut.Code != 0 {
t.Fatalf("forward list failed: code=%d msg=%q", forwardOut.Code, forwardOut.Msg)
}
forwardRows := mustContractSlice(t, forwardOut.Data, "forward list data")
shareOut := requestContractEnvelope(t, router, adminToken, "/api/v1/federation/share/list", nil)
if shareOut.Code != 0 {
t.Fatalf("share list failed: code=%d msg=%q", shareOut.Code, shareOut.Msg)
}
localShareRows := mustContractSlice(t, shareOut.Data, "share list data")
remoteUsageOut := requestContractEnvelope(t, router, adminToken, "/api/v1/federation/share/remote-usage/list", nil)
if remoteUsageOut.Code != 0 {
t.Fatalf("remote usage list failed: code=%d msg=%q", remoteUsageOut.Code, remoteUsageOut.Msg)
}
remoteUsageRows := mustContractSlice(t, remoteUsageOut.Data, "remote usage data")
if len(remoteUsageRows) == 0 {
t.Fatalf("expected non-empty remote usage rows")
}
findForward := func(id int64) map[string]interface{} {
t.Helper()
for _, row := range forwardRows {
m, ok := row.(map[string]interface{})
if !ok {
continue
}
if contractValueAsInt64(m["id"]) == id {
return m
}
}
t.Fatalf("forward %d not found in /forward/list response", id)
return nil
}
flowByShare := make(map[int64]int64)
shareIDsByTunnel := make(map[int64]map[int64]struct{})
for _, row := range remoteUsageRows {
m, ok := row.(map[string]interface{})
if !ok {
continue
}
shareID := contractValueAsInt64(m["shareId"])
currentFlow := contractValueAsInt64(m["currentFlow"])
if shareID > 0 && currentFlow > 0 {
if currentFlow > flowByShare[shareID] {
flowByShare[shareID] = currentFlow
}
}
bindings, _ := m["bindings"].([]interface{})
for _, bindingRaw := range bindings {
binding, ok := bindingRaw.(map[string]interface{})
if !ok {
continue
}
tunnelIDVal := contractValueAsInt64(binding["tunnelId"])
chainType := contractValueAsInt64(binding["chainType"])
if shareID <= 0 || tunnelIDVal <= 0 {
continue
}
if chainType != 1 {
continue
}
setByTunnel, ok := shareIDsByTunnel[tunnelIDVal]
if !ok {
setByTunnel = make(map[int64]struct{})
shareIDsByTunnel[tunnelIDVal] = setByTunnel
}
setByTunnel[shareID] = struct{}{}
}
}
for _, row := range localShareRows {
m, ok := row.(map[string]interface{})
if !ok {
continue
}
shareID := contractValueAsInt64(m["id"])
currentFlow := contractValueAsInt64(m["currentFlow"])
if shareID > 0 && currentFlow > 0 {
if currentFlow > flowByShare[shareID] {
flowByShare[shareID] = currentFlow
}
}
}
targetForward := findForward(forwardID)
parsedByName := contractParseShareIDFromTunnelName(contractValueAsString(targetForward["tunnelName"]))
if parsedByName != 0 {
t.Fatalf("expected custom tunnel name cannot be parsed as Share-*-Port-*, got %d", parsedByName)
}
resolveShareIDForForward := func(forward map[string]interface{}) int64 {
candidates := make(map[int64]struct{})
shareIDFromName := contractParseShareIDFromTunnelName(contractValueAsString(forward["tunnelName"]))
if shareIDFromName > 0 {
candidates[shareIDFromName] = struct{}{}
}
tunnelIDVal := contractValueAsInt64(forward["tunnelId"])
if setByTunnel, ok := shareIDsByTunnel[tunnelIDVal]; ok {
for sid := range setByTunnel {
candidates[sid] = struct{}{}
}
}
var bestShareID int64
bestFlow := int64(0)
for sid := range candidates {
flow := flowByShare[sid]
if flow > bestFlow {
bestFlow = flow
bestShareID = sid
}
}
return bestShareID
}
resolvedShareID := resolveShareIDForForward(targetForward)
if resolvedShareID != remoteShareID {
t.Fatalf("expected resolved shareID=%d via tunnel binding, got %d", remoteShareID, resolvedShareID)
}
forwardCountByShare := make(map[int64]int)
resolvedByForwardID := make(map[int64]int64)
for _, row := range forwardRows {
m, ok := row.(map[string]interface{})
if !ok {
continue
}
fid := contractValueAsInt64(m["id"])
sid := resolveShareIDForForward(m)
if sid > 0 {
resolvedByForwardID[fid] = sid
}
if sid > 0 && flowByShare[sid] > 0 {
forwardCountByShare[sid] = forwardCountByShare[sid] + 1
}
}
directFlow := contractValueAsInt64(targetForward["inFlow"]) + contractValueAsInt64(targetForward["outFlow"])
if directFlow != 0 {
t.Fatalf("fixture expectation failed: directFlow should be 0, got %d", directFlow)
}
shareFlow := flowByShare[resolvedByForwardID[forwardID]]
if shareFlow <= 0 {
t.Fatalf("expected merged share flow > 0 for resolved share %d", resolvedByForwardID[forwardID])
}
count := forwardCountByShare[resolvedByForwardID[forwardID]]
if count <= 0 {
count = 1
}
estimated := shareFlow / int64(count)
if estimated < 1 {
estimated = 1
}
displayFlow := estimated
if displayFlow <= 0 {
t.Fatalf("expected displayFlow > 0 after tunnel-binding-based merge, got %d", displayFlow)
}
if displayFlow != remoteShareFlow {
t.Fatalf("expected displayFlow=%d, got %d", remoteShareFlow, displayFlow)
}
}
func requestContractEnvelope(t *testing.T, router http.Handler, token string, path string, body interface{}) response.R { func requestContractEnvelope(t *testing.T, router http.Handler, token string, path string, body interface{}) response.R {
t.Helper() t.Helper()
+80 -13
View File
@@ -244,17 +244,7 @@ export default function ForwardPage() {
const mergeFederationShareFlow = async ( const mergeFederationShareFlow = async (
forwardsData: Forward[], forwardsData: Forward[],
): Promise<Forward[]> => { ): Promise<Forward[]> => {
const shareIds = new Set<number>(); if (forwardsData.length === 0) {
forwardsData.forEach((forward) => {
const shareId = parseShareIdFromTunnelName(forward.tunnelName || "");
if (shareId) {
shareIds.add(shareId);
}
});
if (shareIds.size === 0) {
return forwardsData; return forwardsData;
} }
@@ -265,6 +255,7 @@ export default function ForwardPage() {
]); ]);
const flowByShare = new Map<number, number>(); const flowByShare = new Map<number, number>();
const shareIdsByTunnel = new Map<number, Set<number>>();
if (usageRes.code === 0 && Array.isArray(usageRes.data)) { if (usageRes.code === 0 && Array.isArray(usageRes.data)) {
usageRes.data.forEach((item: Record<string, unknown>) => { usageRes.data.forEach((item: Record<string, unknown>) => {
@@ -281,6 +272,34 @@ export default function ForwardPage() {
flowByShare.set(shareId, Math.max(prev, currentFlow)); flowByShare.set(shareId, Math.max(prev, currentFlow));
} }
if (Number.isFinite(shareId) && shareId > 0) {
const bindings = Array.isArray(item.bindings)
? (item.bindings as Array<Record<string, unknown>>)
: [];
bindings.forEach((binding) => {
const tunnelId = Number(binding.tunnelId || 0);
const chainType = Number(binding.chainType || 0);
if (!Number.isFinite(tunnelId) || tunnelId <= 0) {
return;
}
if (Number.isFinite(chainType) && chainType !== 1) {
return;
}
let shareSet = shareIdsByTunnel.get(tunnelId);
if (!shareSet) {
shareSet = new Set<number>();
shareIdsByTunnel.set(tunnelId, shareSet);
}
shareSet.add(shareId);
});
}
}); });
} }
@@ -306,10 +325,58 @@ export default function ForwardPage() {
return forwardsData; return forwardsData;
} }
const resolveShareIdForForward = (forward: Forward): number | null => {
const candidates = new Set<number>();
const shareIdFromName = parseShareIdFromTunnelName(forward.tunnelName || "");
if (shareIdFromName) {
candidates.add(shareIdFromName);
}
const tunnelId = Number(forward.tunnelId || 0);
const shareSetByTunnel = shareIdsByTunnel.get(tunnelId);
if (shareSetByTunnel && shareSetByTunnel.size > 0) {
shareSetByTunnel.forEach((shareId) => {
if (Number.isFinite(shareId) && shareId > 0) {
candidates.add(shareId);
}
});
}
if (candidates.size === 0) {
return null;
}
let bestShareId: number | null = null;
let bestFlow = 0;
candidates.forEach((shareId) => {
const shareFlow = flowByShare.get(shareId) || 0;
if (shareFlow > bestFlow) {
bestFlow = shareFlow;
bestShareId = shareId;
}
});
return bestShareId;
};
const resolvedShareByForwardId = new Map<number, number>();
forwardsData.forEach((forward) => {
const shareId = resolveShareIdForForward(forward);
if (shareId) {
resolvedShareByForwardId.set(forward.id, shareId);
}
});
const forwardCountByShare = new Map<number, number>(); const forwardCountByShare = new Map<number, number>();
forwardsData.forEach((forward) => { forwardsData.forEach((forward) => {
const shareId = parseShareIdFromTunnelName(forward.tunnelName || ""); const shareId = resolvedShareByForwardId.get(forward.id) || null;
if (!shareId || !flowByShare.has(shareId)) { if (!shareId || !flowByShare.has(shareId)) {
return; return;
@@ -322,7 +389,7 @@ export default function ForwardPage() {
}); });
return forwardsData.map((forward) => { return forwardsData.map((forward) => {
const shareId = parseShareIdFromTunnelName(forward.tunnelName || ""); const shareId = resolvedShareByForwardId.get(forward.id) || null;
if (!shareId) { if (!shareId) {
return { ...forward, federationShareFlow: undefined }; return { ...forward, federationShareFlow: undefined };