From 9f17d63cdc5ed6fd404bc6b7f27d3d44ea98ea59 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Tue, 24 Feb 2026 23:06:09 +0800 Subject: [PATCH] fix(backend): keep shared federation port-forward services stable --- .../internal/http/handler/federation.go | 129 +++++++++ .../http/handler/federation_share_test.go | 256 ++++++++++++++++++ .../internal/http/handler/flow_policy.go | 100 ++++++- .../handler/flow_policy_federation_test.go | 169 ++++++++++++ go-backend/internal/http/handler/mutations.go | 16 +- go-backend/internal/store/repo/repository.go | 92 +++++++ 6 files changed, 759 insertions(+), 3 deletions(-) diff --git a/go-backend/internal/http/handler/federation.go b/go-backend/internal/http/handler/federation.go index 4f03c99..aa6c33f 100644 --- a/go-backend/internal/http/handler/federation.go +++ b/go-backend/internal/http/handler/federation.go @@ -766,6 +766,37 @@ func (h *Handler) federationTunnelCreate(w http.ResponseWriter, r *http.Request) return } + usedPorts, err := h.repo.ListUsedPortsOnNode(share.NodeID) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + for _, port := range usedPorts { + if port == req.RemotePort { + response.WriteJSON(w, response.Err(403, "Port already in use")) + return + } + } + + runtimeOnPort, err := h.repo.GetActiveForwardPeerShareRuntimeByPort(share.ID, req.RemotePort) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + if runtimeOnPort != nil { + response.WriteJSON(w, response.Err(403, "Port already in use")) + return + } + existsOnNodePort, err := h.repo.ExistsActivePeerShareRuntimeOnNodePort(share.NodeID, req.RemotePort) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + if existsOnNodePort { + response.WriteJSON(w, response.Err(403, "Port already in use")) + return + } + now := time.Now().UnixMilli() tunnelID, err := h.repo.CreateFederationTunnel( fmt.Sprintf("Share-%d-Port-%d", share.ID, req.RemotePort), @@ -780,6 +811,30 @@ func (h *Handler) federationTunnelCreate(w http.ResponseWriter, r *http.Request) return } + runtime := &repo.PeerShareRuntime{ + ShareID: share.ID, + NodeID: share.NodeID, + ReservationID: randomToken(24), + ResourceKey: fmt.Sprintf("federation-forward-%d-%d-%d", share.ID, tunnelID, req.RemotePort), + BindingID: "", + Role: "forward", + ChainName: "", + ServiceName: "", + Protocol: defaultString(req.Protocol, "tcp"), + Strategy: "fifo", + Port: req.RemotePort, + Target: strings.TrimSpace(req.Target), + Applied: 0, + Status: 1, + CreatedTime: now, + UpdatedTime: now, + } + if err := h.repo.CreatePeerShareRuntime(runtime); err != nil { + _ = h.deleteTunnelByID(tunnelID) + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + h.wsServer.SendCommand(share.NodeID, "reload", nil, time.Second*5) response.WriteJSON(w, response.OK(map[string]interface{}{ @@ -1211,9 +1266,83 @@ func (h *Handler) federationRuntimeCommand(w http.ResponseWriter, r *http.Reques response.WriteJSON(w, response.ErrDefault(err.Error())) return } + if strings.EqualFold(cmd, "addservice") || strings.EqualFold(cmd, "updateservice") { + h.bindPeerShareForwardRuntimeServices(share, req.Data) + } response.WriteJSON(w, response.OK(res)) } +type federationForwardServiceBinding struct { + Name string + Port int +} + +func parseFederationForwardServiceBindings(data interface{}) []federationForwardServiceBinding { + dataMap, ok := data.(map[string]interface{}) + if !ok { + return nil + } + rawServices, ok := dataMap["services"] + if !ok { + return nil + } + serviceList, ok := rawServices.([]interface{}) + if !ok { + return nil + } + + bindings := make([]federationForwardServiceBinding, 0, len(serviceList)) + for _, item := range serviceList { + svcMap, ok := item.(map[string]interface{}) + if !ok { + continue + } + name := normalizeForwardRuntimeServiceName(asString(svcMap["name"])) + if name == "" { + continue + } + if _, _, _, ok := parseFlowServiceIDs(name); !ok { + continue + } + addr := strings.TrimSpace(asString(svcMap["addr"])) + if addr == "" { + continue + } + _, portStr, err := net.SplitHostPort(addr) + if err != nil { + continue + } + port, err := strconv.Atoi(portStr) + if err != nil || port <= 0 { + continue + } + bindings = append(bindings, federationForwardServiceBinding{Name: name, Port: port}) + } + return bindings +} + +func (h *Handler) bindPeerShareForwardRuntimeServices(share *repo.PeerShare, data interface{}) { + if h == nil || h.repo == nil || share == nil { + return + } + bindings := parseFederationForwardServiceBindings(data) + if len(bindings) == 0 { + return + } + + now := time.Now().UnixMilli() + for _, binding := range bindings { + runtime, err := h.repo.GetActiveForwardPeerShareRuntimeByPort(share.ID, binding.Port) + if err != nil || runtime == nil || runtime.Status != 1 { + continue + } + if runtime.ServiceName == binding.Name && runtime.Applied == 1 { + continue + } + _ = h.repo.UpdatePeerShareRuntimeServiceName(runtime.ID, binding.Name, now) + } +} + func isFederationRuntimeCommandAllowed(commandType string) bool { switch strings.ToLower(strings.TrimSpace(commandType)) { case "addservice", "updateservice", "deleteservice", "pauseservice", "resumeservice", "addchains", "deletechains", "addlimiters", "deletelimiters", "tcpping", "reload": diff --git a/go-backend/internal/http/handler/federation_share_test.go b/go-backend/internal/http/handler/federation_share_test.go index 1f81bd9..55a8360 100644 --- a/go-backend/internal/http/handler/federation_share_test.go +++ b/go-backend/internal/http/handler/federation_share_test.go @@ -414,6 +414,262 @@ func TestFederationShareResetFlow(t *testing.T) { } } +func TestFederationTunnelCreateCreatesPeerShareRuntime(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = r.Close() }) + + h := New(r, "test-jwt-secret") + now := time.Now().UnixMilli() + + 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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, "federation-forward-node", "federation-forward-secret", "10.90.80.70", "10.90.80.70", "", "24000-24020", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 0, "", "", "").Error; err != nil { + t.Fatalf("insert node: %v", err) + } + nodeID := mustLastInsertID(t, r, "federation-forward-node") + + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "federation-forward-share", + NodeID: nodeID, + Token: "federation-forward-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 24000, + PortRangeEnd: 24020, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share: %v", err) + } + share, err := r.GetPeerShareByToken("federation-forward-token") + if err != nil || share == nil { + t.Fatalf("load share: %v", err) + } + + body, err := json.Marshal(federationTunnelRequest{ + Protocol: "tcp", + RemotePort: 24001, + Target: "1.1.1.1:443", + }) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/api/v1/federation/tunnel/create", bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+share.Token) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + + h.federationTunnelCreate(res, req) + + if res.Code != http.StatusOK { + t.Fatalf("expected status %d, got %d", http.StatusOK, res.Code) + } + + var payload response.R + if err := json.NewDecoder(res.Body).Decode(&payload); err != nil { + t.Fatalf("decode response: %v", err) + } + if payload.Code != 0 { + t.Fatalf("expected response code 0, got %d (%s)", payload.Code, payload.Msg) + } + + runtimeCount := mustQueryInt(t, r, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND port = ? AND status = 1`, share.ID, 24001) + if runtimeCount != 1 { + t.Fatalf("expected 1 runtime row for new federation forward tunnel, got %d", runtimeCount) + } +} + +func TestFederationTunnelCreateRejectsOccupiedPort(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = r.Close() }) + + h := New(r, "test-jwt-secret") + now := time.Now().UnixMilli() + + 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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, "federation-port-check-node", "federation-port-check-secret", "10.91.80.70", "10.91.80.70", "", "24100-24120", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 0, "", "", "").Error; err != nil { + t.Fatalf("insert node: %v", err) + } + nodeID := mustLastInsertID(t, r, "federation-port-check-node") + + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "federation-port-check-share", + NodeID: nodeID, + Token: "federation-port-check-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 24100, + PortRangeEnd: 24120, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share: %v", err) + } + + create := func() response.R { + body, err := json.Marshal(federationTunnelRequest{Protocol: "tcp", RemotePort: 24101, Target: "1.1.1.1:443"}) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/federation/tunnel/create", bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer federation-port-check-token") + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + h.federationTunnelCreate(res, req) + if res.Code != http.StatusOK { + t.Fatalf("expected status %d, got %d", http.StatusOK, res.Code) + } + var payload response.R + if err := json.NewDecoder(res.Body).Decode(&payload); err != nil { + t.Fatalf("decode response: %v", err) + } + return payload + } + + first := create() + if first.Code != 0 { + t.Fatalf("expected first create success, got %d (%s)", first.Code, first.Msg) + } + + second := create() + if second.Code != 403 { + t.Fatalf("expected second create to be rejected with 403, got %d (%s)", second.Code, second.Msg) + } + if second.Msg != "Port already in use" { + t.Fatalf("expected occupied port message, got %q", second.Msg) + } +} + +func TestDeleteTunnelReleasesFederationForwardRuntimeByPort(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = r.Close() }) + + h := New(r, "test-jwt-secret") + now := time.Now().UnixMilli() + + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "delete-forward-share", + NodeID: 1, + Token: "delete-forward-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 25000, + PortRangeEnd: 25020, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share: %v", err) + } + share, err := r.GetPeerShareByToken("delete-forward-token") + if err != nil || share == nil { + t.Fatalf("load share: %v", err) + } + + tunnelName := fmt.Sprintf("Share-%d-Port-%d", share.ID, 25001) + if err := r.DB().Exec(` + INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, 1, tunnelName, 1.0, 1, "tcp", 1, now, now, 1, nil, 0).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + + if err := r.DB().Exec(` + INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, share.ID, share.NodeID, "del-r1", "del-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 25001, "", 1, 1, now, now).Error; err != nil { + t.Fatalf("insert runtime: %v", err) + } + + if err := h.deleteTunnelByID(1); err != nil { + t.Fatalf("delete tunnel: %v", err) + } + + activeCount := mustQueryInt(t, r, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND port = ? AND status = 1`, share.ID, 25001) + if activeCount != 0 { + t.Fatalf("expected runtime released after tunnel delete, active rows=%d", activeCount) + } +} + +func TestBindPeerShareForwardRuntimeServicesOnlyBindsForwardRole(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = r.Close() }) + + h := New(r, "test-jwt-secret") + now := time.Now().UnixMilli() + + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "bind-forward-role-share", + NodeID: 1, + Token: "bind-forward-role-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 26000, + PortRangeEnd: 26020, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share: %v", err) + } + share, err := r.GetPeerShareByToken("bind-forward-role-token") + if err != nil || share == nil { + t.Fatalf("load share: %v", err) + } + + if err := r.DB().Exec(` + INSERT INTO peer_share_runtime(id, share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?), + (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + 1, share.ID, share.NodeID, "bind-r1", "bind-rk1", "", "forward", "", "", "tcp", "fifo", 26001, "", 0, 1, now, now, + 2, share.ID, share.NodeID, "bind-r2", "bind-rk2", "", "middle", "", "", "tcp", "round", 26002, "", 0, 1, now, now, + ).Error; err != nil { + t.Fatalf("insert runtimes: %v", err) + } + + h.bindPeerShareForwardRuntimeServices(share, map[string]interface{}{ + "services": []interface{}{ + map[string]interface{}{"name": "77_2_10_tcp", "addr": "[::]:26001"}, + map[string]interface{}{"name": "88_2_10_tcp", "addr": "[::]:26002"}, + }, + }) + + forwardServiceName := "" + middleServiceName := "" + if err := r.DB().Raw(`SELECT service_name FROM peer_share_runtime WHERE id = 1`).Scan(&forwardServiceName).Error; err != nil { + t.Fatalf("load forward runtime service name: %v", err) + } + if err := r.DB().Raw(`SELECT service_name FROM peer_share_runtime WHERE id = 2`).Scan(&middleServiceName).Error; err != nil { + t.Fatalf("load middle runtime service name: %v", err) + } + + if forwardServiceName != "77_2_10" { + t.Fatalf("expected forward runtime service name bound, got %q", forwardServiceName) + } + if middleServiceName != "" { + t.Fatalf("expected non-forward runtime unchanged, got %q", middleServiceName) + } +} + func TestFederationRemoteUsageList(t *testing.T) { r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db")) if err != nil { diff --git a/go-backend/internal/http/handler/flow_policy.go b/go-backend/internal/http/handler/flow_policy.go index 81c641c..c7a218d 100644 --- a/go-backend/internal/http/handler/flow_policy.go +++ b/go-backend/internal/http/handler/flow_policy.go @@ -40,7 +40,7 @@ func (h *Handler) processFlowItem(item flowItem) { if ok { inFlow, outFlow := h.scaleFlowByTunnel(forwardID, item.D, item.U) _ = h.repo.AddFlow(forwardID, userID, userTunnelID, inFlow, outFlow) - h.processPeerShareFlowFromForward(forwardID, item) + h.processPeerShareFlowFromForward(forwardID, serviceName, item) if userTunnelID > 0 { h.enforceFlowPolicies(userID, userTunnelID) @@ -88,6 +88,28 @@ func parsePeerShareRuntimeServiceID(serviceName string) (int64, bool) { return runtimeID, true } +func parsePeerShareInfoFromFederationTunnelName(tunnelName string) (int64, int, bool) { + tunnelName = strings.TrimSpace(tunnelName) + if !strings.HasPrefix(tunnelName, "Share-") { + return 0, 0, false + } + raw := strings.TrimPrefix(tunnelName, "Share-") + idx := strings.Index(raw, "-Port-") + if idx <= 0 { + return 0, 0, false + } + shareID, err := strconv.ParseInt(raw[:idx], 10, 64) + if err != nil || shareID <= 0 { + return 0, 0, false + } + portValue := strings.TrimSpace(raw[idx+len("-Port-"):]) + port, err := strconv.Atoi(portValue) + if err != nil || port <= 0 { + return 0, 0, false + } + return shareID, port, true +} + func parsePeerShareIDFromFederationTunnelName(tunnelName string) (int64, bool) { tunnelName = strings.TrimSpace(tunnelName) if !strings.HasPrefix(tunnelName, "Share-") { @@ -131,12 +153,15 @@ func (h *Handler) processPeerShareFlow(runtimeID int64, item flowItem) { h.enforcePeerShareFlowLimit(share.ID) } -func (h *Handler) processPeerShareFlowFromForward(forwardID int64, item flowItem) { +func (h *Handler) processPeerShareFlowFromForward(forwardID int64, serviceName string, item flowItem) { if h == nil || h.repo == nil || forwardID <= 0 { return } forward, err := h.getForwardRecord(forwardID) if err != nil || forward == nil { + // Forward not found in local database - might be a federation port-forward + // Try to find by service name in peer_share_runtime + h.processPeerShareFlowByServiceName(serviceName, item) return } tunnel, err := h.getTunnelRecord(forward.TunnelID) @@ -169,6 +194,54 @@ func (h *Handler) processPeerShareFlowFromForward(forwardID int64, item flowItem h.enforcePeerShareFlowLimit(share.ID) } +func normalizeForwardRuntimeServiceName(serviceName string) string { + name := strings.TrimSpace(serviceName) + if strings.HasSuffix(name, "_tcp") { + return strings.TrimSuffix(name, "_tcp") + } + if strings.HasSuffix(name, "_udp") { + return strings.TrimSuffix(name, "_udp") + } + return name +} + +func (h *Handler) processPeerShareFlowByServiceName(serviceName string, item flowItem) { + if h == nil || h.repo == nil || strings.TrimSpace(serviceName) == "" { + return + } + + delta := item.D + item.U + if delta <= 0 { + return + } + + normalized := normalizeForwardRuntimeServiceName(serviceName) + runtimes, err := h.repo.ListActiveForwardPeerShareRuntimesByServiceName(normalized) + if err != nil { + return + } + if len(runtimes) == 0 && normalized != serviceName { + runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(serviceName) + if err != nil { + return + } + } + if len(runtimes) != 1 { + return + } + runtime := runtimes[0] + + _ = h.repo.AddPeerShareCurrentFlow(runtime.ShareID, delta) + + matchedShare, err := h.repo.GetPeerShare(runtime.ShareID) + if err != nil || matchedShare == nil { + return + } + if isPeerShareFlowExceeded(matchedShare) { + h.enforcePeerShareFlowLimit(matchedShare.ID) + } +} + func (h *Handler) enforcePeerShareFlowLimit(shareID int64) { if h == nil || h.repo == nil || shareID <= 0 { return @@ -327,11 +400,34 @@ func (h *Handler) cleanNodeConfigs(nodeID int64, rawConfig string) { } func (h *Handler) cleanOrphanedServices(nodeID int64, services []namedConfigItem) { + runtimeServiceNames, err := h.repo.ListActiveForwardPeerShareRuntimeServiceNamesByNode(nodeID) + if err != nil { + return + } + runtimeServiceSet := make(map[string]struct{}, len(runtimeServiceNames)) + for _, serviceName := range runtimeServiceNames { + serviceName = strings.TrimSpace(serviceName) + if serviceName == "" { + continue + } + runtimeServiceSet[serviceName] = struct{}{} + } + for _, item := range services { name := strings.TrimSpace(item.Name) if name == "" || name == "web_api" { continue } + if strings.HasPrefix(name, "fed_svc_") { + continue + } + normalizedName := normalizeForwardRuntimeServiceName(name) + if _, ok := runtimeServiceSet[normalizedName]; ok { + continue + } + if _, ok := runtimeServiceSet[name]; ok { + continue + } parts := strings.Split(name, "_") if len(parts) >= 3 { diff --git a/go-backend/internal/http/handler/flow_policy_federation_test.go b/go-backend/internal/http/handler/flow_policy_federation_test.go index 81e8dcf..0e38b4b 100644 --- a/go-backend/internal/http/handler/flow_policy_federation_test.go +++ b/go-backend/internal/http/handler/flow_policy_federation_test.go @@ -130,3 +130,172 @@ func TestProcessFlowItemTracksPeerShareFlowForFederationPortForward(t *testing.T t.Fatalf("expected current_flow=200, got %d", updatedShare.CurrentFlow) } } + +func TestProcessFlowItemTracksPeerShareFlowByForwardServiceName(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel-forward-service.db")) + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "forward-service-share", + NodeID: 1, + Token: "forward-service-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 31000, + PortRangeEnd: 31010, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create peer share: %v", err) + } + share, err := r.GetPeerShareByToken("forward-service-token") + if err != nil || share == nil { + t.Fatalf("load peer share: %v", err) + } + + if err := r.DB().Exec(` + INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, share.ID, share.NodeID, "svc-r1", "svc-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 31001, "", 1, 1, now, now).Error; err != nil { + t.Fatalf("insert peer_share_runtime: %v", err) + } + + h := &Handler{repo: r} + h.processFlowItem(flowItem{N: "20_2_10_tcp", U: 120, D: 80}) + + updatedShare, err := r.GetPeerShare(share.ID) + if err != nil || updatedShare == nil { + t.Fatalf("reload share: %v", err) + } + if updatedShare.CurrentFlow != 200 { + t.Fatalf("expected current_flow=200, got %d", updatedShare.CurrentFlow) + } +} + +func TestProcessFlowItemSkipsPeerShareFlowWhenServiceNameIsAmbiguous(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel-forward-ambiguous.db")) + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "ambiguous-share-a", + NodeID: 1, + Token: "ambiguous-token-a", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 31100, + PortRangeEnd: 31110, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share A: %v", err) + } + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "ambiguous-share-b", + NodeID: 1, + Token: "ambiguous-token-b", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 31200, + PortRangeEnd: 31210, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share B: %v", err) + } + shareA, _ := r.GetPeerShareByToken("ambiguous-token-a") + shareB, _ := r.GetPeerShareByToken("ambiguous-token-b") + + if err := r.DB().Exec(` + INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?), + (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + shareA.ID, 1, "amb-r1", "amb-rk1", "", "forward", "", "99_2_10", "tcp", "fifo", 31101, "", 1, 1, now, now, + shareB.ID, 1, "amb-r2", "amb-rk2", "", "forward", "", "99_2_10", "tcp", "fifo", 31201, "", 1, 1, now, now, + ).Error; err != nil { + t.Fatalf("insert ambiguous runtimes: %v", err) + } + + h := &Handler{repo: r} + h.processFlowItem(flowItem{N: "99_2_10_tcp", U: 120, D: 80}) + + updatedA, _ := r.GetPeerShare(shareA.ID) + updatedB, _ := r.GetPeerShare(shareB.ID) + if updatedA.CurrentFlow != 0 || updatedB.CurrentFlow != 0 { + t.Fatalf("expected ambiguous service flow to be skipped, got shareA=%d shareB=%d", updatedA.CurrentFlow, updatedB.CurrentFlow) + } +} + +func TestCleanOrphanedServicesSkipsActiveSharedForwardRuntimeServices(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel-cleanup-runtime.db")) + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "cleanup-runtime-share", + NodeID: 1, + Token: "cleanup-runtime-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 31300, + PortRangeEnd: 31310, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create peer share: %v", err) + } + share, err := r.GetPeerShareByToken("cleanup-runtime-token") + if err != nil || share == nil { + t.Fatalf("load peer share: %v", err) + } + + if err := r.DB().Exec(` + INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, share.ID, share.NodeID, "cleanup-r1", "cleanup-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 31301, "", 1, 1, now, now).Error; err != nil { + t.Fatalf("insert peer_share_runtime: %v", err) + } + + h := &Handler{repo: r} + + defer func() { + if rec := recover(); rec != nil { + t.Fatalf("cleanOrphanedServices should skip active shared runtime service; got panic: %v", rec) + } + }() + + h.cleanOrphanedServices(share.NodeID, []namedConfigItem{{Name: "20_2_10_tcp"}}) +} + +func TestCleanOrphanedServicesSkipsFederationServicePrefix(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel-cleanup-fed-svc.db")) + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + h := &Handler{repo: r} + + defer func() { + if rec := recover(); rec != nil { + t.Fatalf("cleanOrphanedServices should skip fed_svc_ service names; got panic: %v", rec) + } + }() + + h.cleanOrphanedServices(1, []namedConfigItem{{Name: "fed_svc_999_tcp"}}) +} diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 71d3eca..c0132d5 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -2794,7 +2794,21 @@ func (h *Handler) deleteNodeByID(id int64) error { } func (h *Handler) deleteTunnelByID(id int64) error { - return h.repo.DeleteTunnelCascade(id) + if h == nil || h.repo == nil { + return errors.New("repository not initialized") + } + + tunnelName, _ := h.repo.GetTunnelName(id) + if err := h.repo.DeleteTunnelCascade(id); err != nil { + return err + } + + shareID, port, ok := parsePeerShareInfoFromFederationTunnelName(tunnelName) + if !ok { + return nil + } + + return h.repo.MarkPeerShareRuntimeReleasedByPort(shareID, port, time.Now().UnixMilli()) } func (h *Handler) deleteForwardByID(id int64) error { diff --git a/go-backend/internal/store/repo/repository.go b/go-backend/internal/store/repo/repository.go index bbdf042..e378bfb 100644 --- a/go-backend/internal/store/repo/repository.go +++ b/go-backend/internal/store/repo/repository.go @@ -1270,6 +1270,98 @@ func (r *Repository) ListActivePeerShareRuntimePorts(shareID int64, nodeID int64 return ports, nil } +func (r *Repository) ListActiveForwardPeerShareRuntimesByServiceName(serviceName string) ([]model.PeerShareRuntime, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + var items []model.PeerShareRuntime + err := r.db.Where("service_name = ? AND status = 1 AND role = ?", serviceName, "forward"). + Order("id ASC"). + Find(&items).Error + if err != nil { + return nil, err + } + if items == nil { + items = make([]model.PeerShareRuntime, 0) + } + return items, nil +} + +func (r *Repository) ListActiveForwardPeerShareRuntimeServiceNamesByNode(nodeID int64) ([]string, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + var names []string + err := r.db.Model(&model.PeerShareRuntime{}). + Where("node_id = ? AND status = 1 AND role = ? AND service_name <> ''", nodeID, "forward"). + Pluck("service_name", &names).Error + if err != nil { + return nil, err + } + if names == nil { + names = make([]string, 0) + } + return names, nil +} + +func (r *Repository) GetActiveForwardPeerShareRuntimeByPort(shareID int64, port int) (*model.PeerShareRuntime, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + var item model.PeerShareRuntime + err := r.db.Where("share_id = ? AND port = ? AND status = 1 AND role = ?", shareID, port, "forward").First(&item).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &item, nil +} + +func (r *Repository) ExistsActivePeerShareRuntimeOnNodePort(nodeID int64, port int) (bool, error) { + if r == nil || r.db == nil { + return false, errors.New("repository not initialized") + } + var count int64 + err := r.db.Model(&model.PeerShareRuntime{}). + Where("node_id = ? AND port = ? AND status = 1", nodeID, port). + Count(&count).Error + if err != nil { + return false, err + } + return count > 0, nil +} + +func (r *Repository) UpdatePeerShareRuntimeServiceName(id int64, serviceName string, updatedTime int64) error { + if r == nil || r.db == nil { + return errors.New("repository not initialized") + } + return r.db.Model(&model.PeerShareRuntime{}).Where("id = ?", id).Updates(map[string]interface{}{ + "service_name": serviceName, + "applied": 1, + "updated_time": updatedTime, + }).Error +} + +func (r *Repository) MarkPeerShareRuntimeReleasedByPort(shareID int64, port int, updatedTime int64) error { + if r == nil || r.db == nil { + return errors.New("repository not initialized") + } + if shareID <= 0 || port <= 0 { + return nil + } + if updatedTime <= 0 { + updatedTime = unixMilliNow() + } + return r.db.Model(&model.PeerShareRuntime{}).Where("share_id = ? AND port = ? AND status = 1", shareID, port).Updates(map[string]interface{}{ + "status": 0, + "applied": 0, + "service_name": "", + "updated_time": updatedTime, + }).Error +} + // ─── FederationTunnelBinding ───────────────────────────────────────── func (r *Repository) UpsertFederationTunnelBinding(item *model.FederationTunnelBinding) error {