fix(federation): clean up runtimes on share delete and sync live traffic from provider

Co-authored-by: Sisyphus <sisyphus@opencode.ai>
2026-09-29 16:06:36 +08:00 · 2026-02-10 11:57:52 +00:00
parent 919339cdb1
commit 89ab12dcdf
2 changed files with 211 additions and 8 deletions
@@ -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)
+	}
+}
@@ -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 {