From 89ab12dcdf7c3d7a05d88cc607152c9bd6af9c03 Mon Sep 17 00:00:00 2001 From: sagit Date: Tue, 10 Feb 2026 11:57:52 +0000 Subject: [PATCH] fix(federation): clean up runtimes on share delete and sync live traffic from provider Co-authored-by: Sisyphus --- .../internal/http/handler/federation.go | 77 +++++++++- .../http/handler/federation_share_test.go | 142 ++++++++++++++++++ 2 files changed, 211 insertions(+), 8 deletions(-) diff --git a/go-backend/internal/http/handler/federation.go b/go-backend/internal/http/handler/federation.go index 874b837..e3bea0d 100644 --- a/go-backend/internal/http/handler/federation.go +++ b/go-backend/internal/http/handler/federation.go @@ -117,9 +117,11 @@ type remoteUsageNodeItem struct { PortRangeEnd int `json:"portRangeEnd"` MaxBandwidth int64 `json:"maxBandwidth"` CurrentFlow int64 `json:"currentFlow"` + ExpiryTime int64 `json:"expiryTime"` UsedPorts []int `json:"usedPorts"` Bindings []remoteUsageBindingItem `json:"bindings"` ActiveBindingNum int `json:"activeBindingNum"` + SyncError string `json:"syncError,omitempty"` } func (h *Handler) federationShareList(w http.ResponseWriter, r *http.Request) { @@ -279,6 +281,8 @@ func (h *Handler) federationShareDelete(w http.ResponseWriter, r *http.Request) return } + h.cleanupPeerShareRuntimes(req.ID) + if err := h.repo.DeletePeerShare(req.ID); err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return @@ -328,7 +332,7 @@ func (h *Handler) federationRemoteUsageList(w http.ResponseWriter, r *http.Reque } rows, err := h.repo.DB().Query(` - SELECT id, name, remote_url, remote_config + SELECT id, name, remote_url, remote_token, remote_config FROM node WHERE is_remote = 1 ORDER BY id DESC @@ -339,20 +343,51 @@ func (h *Handler) federationRemoteUsageList(w http.ResponseWriter, r *http.Reque } defer rows.Close() + fc := client.NewFederationClient() + localDomain := h.federationLocalDomain() + items := make([]remoteUsageNodeItem, 0) for rows.Next() { var ( nodeID int64 nodeName string remoteURL sql.NullString + remoteToken sql.NullString remoteConfig sql.NullString ) - if err := rows.Scan(&nodeID, &nodeName, &remoteURL, &remoteConfig); err != nil { + if err := rows.Scan(&nodeID, &nodeName, &remoteURL, &remoteToken, &remoteConfig); err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return } - shareID, maxBandwidth, currentFlow, portRangeStart, portRangeEnd := parseRemoteShareUsageConfig(remoteConfig.String) + shareID, maxBandwidth, currentFlow, expiryTime, portRangeStart, portRangeEnd := parseRemoteShareUsageConfig(remoteConfig.String) + + var syncError string + url := strings.TrimSpace(remoteURL.String) + token := strings.TrimSpace(remoteToken.String) + if url != "" && token != "" { + info, connectErr := fc.Connect(url, token, localDomain) + if connectErr != nil { + syncError = connectErr.Error() + } else if info != nil { + shareID = info.ShareID + maxBandwidth = info.MaxBandwidth + currentFlow = info.CurrentFlow + expiryTime = info.ExpiryTime + portRangeStart = info.PortRangeStart + portRangeEnd = info.PortRangeEnd + + configData, _ := json.Marshal(map[string]interface{}{ + "shareId": info.ShareID, + "maxBandwidth": info.MaxBandwidth, + "currentFlow": info.CurrentFlow, + "expiryTime": info.ExpiryTime, + "portRangeStart": info.PortRangeStart, + "portRangeEnd": info.PortRangeEnd, + }) + _, _ = h.repo.DB().Exec(`UPDATE node SET remote_config = ? WHERE id = ?`, string(configData), nodeID) + } + } bindingRows, err := h.repo.DB().Query(` SELECT fb.id, fb.tunnel_id, COALESCE(t.name, ''), fb.chain_type, fb.hop_inx, fb.allocated_port, fb.resource_key, fb.remote_binding_id, fb.updated_time @@ -396,15 +431,17 @@ func (h *Handler) federationRemoteUsageList(w http.ResponseWriter, r *http.Reque items = append(items, remoteUsageNodeItem{ NodeID: nodeID, NodeName: nodeName, - RemoteURL: strings.TrimSpace(remoteURL.String), + RemoteURL: url, ShareID: shareID, PortRangeStart: portRangeStart, PortRangeEnd: portRangeEnd, MaxBandwidth: maxBandwidth, CurrentFlow: currentFlow, + ExpiryTime: expiryTime, UsedPorts: usedPorts, Bindings: bindings, ActiveBindingNum: len(bindings), + SyncError: syncError, }) } if err := rows.Err(); err != nil { @@ -415,23 +452,24 @@ func (h *Handler) federationRemoteUsageList(w http.ResponseWriter, r *http.Reque response.WriteJSON(w, response.OK(items)) } -func parseRemoteShareUsageConfig(raw string) (int64, int64, int64, int, int) { +func parseRemoteShareUsageConfig(raw string) (int64, int64, int64, int64, int, int) { raw = strings.TrimSpace(raw) if raw == "" { - return 0, 0, 0, 0, 0 + return 0, 0, 0, 0, 0, 0 } var cfg map[string]interface{} if err := json.Unmarshal([]byte(raw), &cfg); err != nil { - return 0, 0, 0, 0, 0 + return 0, 0, 0, 0, 0, 0 } shareID := asInt64(cfg["shareId"], 0) maxBandwidth := asInt64(cfg["maxBandwidth"], 0) currentFlow := asInt64(cfg["currentFlow"], 0) + expiryTime := asInt64(cfg["expiryTime"], 0) portRangeStart := int(asInt64(cfg["portRangeStart"], 0)) portRangeEnd := int(asInt64(cfg["portRangeEnd"], 0)) - return shareID, maxBandwidth, currentFlow, portRangeStart, portRangeEnd + return shareID, maxBandwidth, currentFlow, expiryTime, portRangeStart, portRangeEnd } func (h *Handler) nodeImport(w http.ResponseWriter, r *http.Request) { @@ -1278,3 +1316,26 @@ func isPeerIPAllowed(clientIP net.IP, whitelist string) bool { return false } + +func (h *Handler) cleanupPeerShareRuntimes(shareID int64) { + if h == nil || h.repo == nil || shareID <= 0 { + return + } + runtimes, err := h.repo.ListActivePeerShareRuntimesByShareID(shareID) + if err != nil || len(runtimes) == 0 { + return + } + + now := time.Now().UnixMilli() + for _, runtime := range runtimes { + if h.wsServer != nil && runtime.Applied == 1 { + if strings.TrimSpace(runtime.ServiceName) != "" { + _, _ = h.sendNodeCommand(runtime.NodeID, "DeleteService", map[string]interface{}{"services": []string{runtime.ServiceName}}, false, true) + } + if strings.TrimSpace(runtime.Role) == "middle" && strings.TrimSpace(runtime.ChainName) != "" { + _, _ = h.sendNodeCommand(runtime.NodeID, "DeleteChains", map[string]interface{}{"chain": runtime.ChainName}, false, true) + } + } + _ = h.repo.MarkPeerShareRuntimeReleased(runtime.ID, now) + } +} diff --git a/go-backend/internal/http/handler/federation_share_test.go b/go-backend/internal/http/handler/federation_share_test.go index ad29190..37cd8a1 100644 --- a/go-backend/internal/http/handler/federation_share_test.go +++ b/go-backend/internal/http/handler/federation_share_test.go @@ -237,6 +237,148 @@ func TestFederationShareListIncludesRemoteUsedPorts(t *testing.T) { } } +func TestFederationShareDeleteCleansUpRuntimes(t *testing.T) { + repo, err := sqlite.Open(filepath.Join(t.TempDir(), "panel.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = repo.Close() }) + + h := New(repo, "test-jwt-secret") + now := time.Now().UnixMilli() + + if err := repo.CreatePeerShare(&sqlite.PeerShare{ + Name: "delete-cleanup-share", + NodeID: 99, + Token: "delete-cleanup-token", + MaxBandwidth: 4096, + PortRangeStart: 40000, + PortRangeEnd: 40010, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create peer share: %v", err) + } + + share, err := repo.GetPeerShareByToken("delete-cleanup-token") + if err != nil || share == nil { + t.Fatalf("load peer share: %v", err) + } + + if _, err := repo.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, 99, "dc-r1", "dc-rk1", "dc-b1", "exit", "", "fed_svc_dc1", "tls", "round", 40001, "", 1, 1, now, now, + share.ID, 99, "dc-r2", "dc-rk2", "dc-b2", "middle", "fed_chain_dc2", "fed_svc_dc2", "tls", "round", 40002, "", 1, 1, now, now, + ); err != nil { + t.Fatalf("insert peer_share_runtime rows: %v", err) + } + + var runtimeCount int + if err := repo.DB().QueryRow(`SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1`, share.ID).Scan(&runtimeCount); err != nil { + t.Fatalf("count active runtimes before: %v", err) + } + if runtimeCount != 2 { + t.Fatalf("expected 2 active runtimes before delete, got %d", runtimeCount) + } + + body, err := json.Marshal(deletePeerShareRequest{ID: share.ID}) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/federation/share/delete", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + + h.federationShareDelete(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) + } + + var shareCount int + if err := repo.DB().QueryRow(`SELECT COUNT(1) FROM peer_share WHERE id = ?`, share.ID).Scan(&shareCount); err != nil { + t.Fatalf("count peer_share after: %v", err) + } + if shareCount != 0 { + t.Fatalf("expected peer_share deleted, got %d rows", shareCount) + } + + var runtimeCountAfter int + if err := repo.DB().QueryRow(`SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ?`, share.ID).Scan(&runtimeCountAfter); err != nil { + t.Fatalf("count peer_share_runtime after: %v", err) + } + if runtimeCountAfter != 0 { + t.Fatalf("expected all peer_share_runtime rows deleted, got %d", runtimeCountAfter) + } +} + +func TestFederationRemoteUsageListSyncErrorFallback(t *testing.T) { + repo, err := sqlite.Open(filepath.Join(t.TempDir(), "panel.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = repo.Close() }) + + h := New(repo, "test-jwt-secret") + now := time.Now().UnixMilli() + + if _, err := repo.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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, "sync-error-node", "sync-error-secret", "10.50.60.70", "10.50.60.70", "", "32000-32010", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 1, "http://unreachable.invalid:9999", "bad-token", `{"shareId":42,"maxBandwidth":5368709120,"currentFlow":999999,"portRangeStart":32000,"portRangeEnd":32010}`); err != nil { + t.Fatalf("insert remote node: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/api/v1/federation/share/remote-usage/list", nil) + res := httptest.NewRecorder() + h.federationRemoteUsageList(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) + } + + rows, ok := payload.Data.([]interface{}) + if !ok || len(rows) == 0 { + t.Fatalf("expected non-empty usage list, got %T", payload.Data) + } + + first, ok := rows[0].(map[string]interface{}) + if !ok { + t.Fatalf("expected row map, got %T", rows[0]) + } + + if int64(first["shareId"].(float64)) != 42 { + t.Fatalf("expected stale shareId=42 on sync failure, got %v", first["shareId"]) + } + if int64(first["currentFlow"].(float64)) != 999999 { + t.Fatalf("expected stale currentFlow=999999 on sync failure, got %v", first["currentFlow"]) + } + + syncErr, _ := first["syncError"].(string) + if syncErr == "" { + t.Fatalf("expected non-empty syncError field on unreachable provider") + } +} + func TestFederationShareResetFlow(t *testing.T) { repo, err := sqlite.Open(filepath.Join(t.TempDir(), "panel.db")) if err != nil {