fix(backend): clean stale forward runtimes on entry updates (#307)

Entire-Checkpoint: 6a6b91fb5f0c

go-backend/internal/http/handler/control_plane.go

@@ -312,6 +312,14 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method
 			return warnings, fmt.Errorf("节点 %s 下发失败: %w", node.Name, err)
 		}
 	}
+
+	// Keep paused forwards paused after UpdateService/AddService, since agent-side UpdateService
+	// always restarts services.
+	if forward.Status != 1 {
+		if err := h.controlForwardServices(forward, "PauseService", false); err != nil {
+			return warnings, err
+		}
+	}
 	return warnings, nil
 }
 
@@ -1620,6 +1628,13 @@ func processServerAddress(serverAddr string) string {
 	if strings.HasPrefix(serverAddr, "[") {
 		return serverAddr
 	}
+	// If the input is a bare IPv6 host (no port), bracket it.
+	// IPv6-with-port must be provided in bracket form: [::1]:443.
+	if looksLikeIPv6(serverAddr) {
+		if ip := net.ParseIP(serverAddr); ip != nil && ip.To4() == nil {
+			return "[" + serverAddr + "]"
+		}
+	}
 
 	idx := strings.LastIndex(serverAddr, ":")
 	if idx < 0 {

go-backend/internal/http/handler/flow_policy.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"encoding/json"
+	"errors"
 	"log"
 	"strconv"
 	"strings"
@@ -327,6 +328,67 @@ func (h *Handler) enforceFlowPolicies(userID int64, userTunnelID int64) {
 	}
 }
 
+func (h *Handler) ensureUserTunnelForwardAllowed(userID int64, tunnelID int64, now int64) error {
+	if h == nil || h.repo == nil {
+		return errors.New("invalid flow policy context")
+	}
+	if userID <= 0 || tunnelID <= 0 {
+		return nil
+	}
+
+	user, err := h.repo.GetUserByID(userID)
+	if err != nil {
+		return err
+	}
+	if user == nil {
+		return errors.New("用户不存在")
+	}
+
+	if user.Status != 1 {
+		return errors.New("账号已禁用")
+	}
+	if user.ExpTime > 0 && user.ExpTime <= now {
+		return errors.New("账号已过期")
+	}
+
+	flowLimit := user.Flow * bytesPerGB
+	current := user.InFlow + user.OutFlow
+	if flowLimit < current {
+		return errors.New("流量已超额，禁止开启转发")
+	}
+
+	userTunnelID, _, _, err := h.resolveUserTunnelAndLimiter(userID, tunnelID)
+	if err != nil {
+		return err
+	}
+	if userTunnelID <= 0 {
+		return nil
+	}
+
+	policy, err := h.getUserTunnelPolicy(userTunnelID)
+	if err != nil {
+		return err
+	}
+	if policy == nil {
+		return nil
+	}
+
+	if policy.Status != 1 {
+		return errors.New("该隧道已禁用")
+	}
+	if policy.ExpTime > 0 && policy.ExpTime <= now {
+		return errors.New("该隧道已过期")
+	}
+
+	utFlowLimit := policy.Flow * bytesPerGB
+	utCurrent := policy.InFlow + policy.OutFlow
+	if utCurrent >= utFlowLimit {
+		return errors.New("该隧道流量已超额，禁止开启转发")
+	}
+
+	return nil
+}
+
 func (h *Handler) shouldPauseUser(userID int64, now int64) bool {
 	user, err := h.repo.GetUserByID(userID)
 	if err != nil || user == nil {

go-backend/internal/http/handler/mutations.go

@@ -684,6 +684,7 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) {
 		response.WriteJSON(w, response.ErrDefault("隧道ID不能为空"))
 		return
 	}
+	oldEntryNodeIDs, _ := h.tunnelEntryNodeIDs(id)
 
 	h.cleanupTunnelRuntime(id)
 	h.cleanupFederationRuntime(id)
@@ -755,6 +756,12 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	newEntryNodeIDs, _ := h.tunnelEntryNodeIDs(id)
+	if !sameInt64Set(oldEntryNodeIDs, newEntryNodeIDs) {
+		h.cleanupTunnelForwardRuntimesOnRemovedEntryNodes(id, oldEntryNodeIDs, newEntryNodeIDs)
+		h.syncTunnelForwardsEntryPorts(id, newEntryNodeIDs)
+	}
+
 	if typeVal == 2 {
 		createdChains, createdServices, applyErr := h.applyTunnelRuntime(runtimeState)
 		if applyErr != nil {
@@ -779,6 +786,146 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) {
 	response.WriteJSON(w, response.OKEmpty())
 }
 
+func sameInt64Set(a, b []int64) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	if len(a) == 0 {
+		return true
+	}
+	m := make(map[int64]int, len(a))
+	for _, v := range a {
+		m[v]++
+	}
+	for _, v := range b {
+		c, ok := m[v]
+		if !ok || c <= 0 {
+			return false
+		}
+		if c == 1 {
+			delete(m, v)
+			continue
+		}
+		m[v] = c - 1
+	}
+	return len(m) == 0
+}
+
+func pickForwardPortFromRecords(ports []forwardPortRecord) int {
+	min := 0
+	for _, fp := range ports {
+		if fp.Port <= 0 {
+			continue
+		}
+		if min == 0 || fp.Port < min {
+			min = fp.Port
+		}
+	}
+	return min
+}
+
+func uniqueInt64s(input []int64) []int64 {
+	if len(input) <= 1 {
+		return input
+	}
+	seen := make(map[int64]struct{}, len(input))
+	out := make([]int64, 0, len(input))
+	for _, v := range input {
+		if _, ok := seen[v]; ok {
+			continue
+		}
+		seen[v] = struct{}{}
+		out = append(out, v)
+	}
+	return out
+}
+
+func diffInt64s(base, subtract []int64) []int64 {
+	if len(base) == 0 {
+		return nil
+	}
+	seen := make(map[int64]struct{}, len(subtract))
+	for _, v := range subtract {
+		seen[v] = struct{}{}
+	}
+	out := make([]int64, 0, len(base))
+	for _, v := range base {
+		if _, ok := seen[v]; ok {
+			continue
+		}
+		out = append(out, v)
+	}
+	return uniqueInt64s(out)
+}
+
+func (h *Handler) cleanupTunnelForwardRuntimesOnRemovedEntryNodes(tunnelID int64, oldEntryNodeIDs, newEntryNodeIDs []int64) {
+	if h == nil || h.repo == nil || tunnelID <= 0 {
+		return
+	}
+
+	removedNodeIDs := diffInt64s(oldEntryNodeIDs, newEntryNodeIDs)
+	if len(removedNodeIDs) == 0 {
+		return
+	}
+
+	forwards, err := h.listForwardsByTunnel(tunnelID)
+	if err != nil || len(forwards) == 0 {
+		return
+	}
+
+	for i := range forwards {
+		f := &forwards[i]
+		if f == nil {
+			continue
+		}
+		for _, nodeID := range removedNodeIDs {
+			_ = h.deleteForwardServicesOnNode(f, nodeID)
+		}
+	}
+}
+
+func (h *Handler) syncTunnelForwardsEntryPorts(tunnelID int64, entryNodeIDs []int64) {
+	if h == nil || h.repo == nil || tunnelID <= 0 {
+		return
+	}
+	entryNodeIDs = uniqueInt64s(entryNodeIDs)
+	if len(entryNodeIDs) == 0 {
+		return
+	}
+
+	forwards, err := h.listForwardsByTunnel(tunnelID)
+	if err != nil || len(forwards) == 0 {
+		return
+	}
+
+	allowInIP := len(entryNodeIDs) == 1
+	for i := range forwards {
+		f := &forwards[i]
+		if f == nil {
+			continue
+		}
+		oldPorts, err := h.listForwardPorts(f.ID)
+		if err != nil {
+			continue
+		}
+		port := pickForwardPortFromRecords(oldPorts)
+		if port <= 0 {
+			continue
+		}
+
+		var entries []forwardPortReplaceEntry
+		if allowInIP {
+			entries = buildForwardPortEntriesWithPreservedInIP(entryNodeIDs, oldPorts, port)
+		} else {
+			entries = make([]forwardPortReplaceEntry, 0, len(entryNodeIDs))
+			for _, nid := range entryNodeIDs {
+				entries = append(entries, forwardPortReplaceEntry{NodeID: nid, Port: port, InIP: ""})
+			}
+		}
+		_ = h.repo.ReplaceForwardPorts(f.ID, entries)
+	}
+}
+
 func (h *Handler) tunnelDelete(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		response.WriteJSON(w, response.ErrDefault("请求失败"))
@@ -1155,6 +1302,10 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
 		response.WriteJSON(w, response.ErrDefault("隧道已禁用，无法创建转发"))
 		return
 	}
+	if err := h.ensureUserTunnelForwardAllowed(userID, tunnelID, time.Now().UnixMilli()); err != nil {
+		response.WriteJSON(w, response.ErrDefault(err.Error()))
+		return
+	}
 	name := asString(req["name"])
 	remoteAddr := asString(req["remoteAddr"])
 	if name == "" || remoteAddr == "" {
@@ -1444,11 +1595,16 @@ func (h *Handler) forwardResume(w http.ResponseWriter, r *http.Request) {
 		response.WriteJSON(w, response.Err(-2, err.Error()))
 		return
 	}
+	now := time.Now().UnixMilli()
+	if err := h.ensureUserTunnelForwardAllowed(forward.UserID, forward.TunnelID, now); err != nil {
+		response.WriteJSON(w, response.ErrDefault(err.Error()))
+		return
+	}
 	if err := h.controlForwardServices(forward, "ResumeService", false); err != nil {
 		response.WriteJSON(w, response.ErrDefault(err.Error()))
 		return
 	}
-	_ = h.repo.UpdateForwardStatus(id, 1, time.Now().UnixMilli())
+	_ = h.repo.UpdateForwardStatus(id, 1, now)
 	response.WriteJSON(w, response.OKEmpty())
 }
 
@@ -1571,17 +1727,22 @@ func (h *Handler) forwardBatchResume(w http.ResponseWriter, r *http.Request) {
 	}
 	s := 0
 	f := 0
+	now := time.Now().UnixMilli()
 	for _, id := range ids {
 		forward, accessErr := h.ensureForwardAccessByActor(actorUserID, actorRole, id)
 		if accessErr != nil {
 			f++
 			continue
 		}
+		if err := h.ensureUserTunnelForwardAllowed(forward.UserID, forward.TunnelID, now); err != nil {
+			f++
+			continue
+		}
 		if err := h.controlForwardServices(forward, "ResumeService", false); err != nil {
 			f++
 			continue
 		}
-		if err := h.repo.UpdateForwardStatus(id, 1, time.Now().UnixMilli()); err != nil {
+		if err := h.repo.UpdateForwardStatus(id, 1, now); err != nil {
 			f++
 		} else {
 			s++

go-backend/tests/contract/flow_limit_forward_enable_contract_test.go

@@ -0,0 +1,201 @@
+package contract_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"go-backend/internal/auth"
+	"go-backend/internal/http/response"
+)
+
+const contractBytesPerGB int64 = 1024 * 1024 * 1024
+
+func TestForwardResumeBlockedWhenUserFlowExceeded(t *testing.T) {
+	secret := "contract-jwt-secret"
+	router, repo := setupContractRouter(t, secret)
+	now := time.Now().UnixMilli()
+
+	userID := int64(2)
+	tunnelID := int64(1)
+	forwardID := int64(1)
+
+	flowGB := int64(120)
+	used := flowGB*contractBytesPerGB + 1
+
+	if err := repo.DB().Exec(`
+		INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+		VALUES(?, 'flow_user', 'pwd', 1, 2727251700000, ?, ?, 0, 1, 99999, ?, ?, 1)
+	`, userID, flowGB, used, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+
+	if err := repo.DB().Exec(`
+		INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, 'flow_tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0)
+	`, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	if err := repo.DB().Exec(`
+		INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+		VALUES(10, ?, ?, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)
+	`, userID, tunnelID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+	if err := repo.DB().Exec(`
+		INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
+		VALUES(?, ?, 'flow_user', 'flow_forward', ?, '8.8.8.8:53', 'fifo', 0, 0, ?, ?, 0, 0)
+	`, forwardID, userID, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+
+	token, err := auth.GenerateToken(userID, "flow_user", 1, secret)
+	if err != nil {
+		t.Fatalf("generate token: %v", err)
+	}
+
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/resume", bytes.NewBufferString(`{"id":1}`))
+	req.Header.Set("Authorization", token)
+	req.Header.Set("Content-Type", "application/json")
+	res := httptest.NewRecorder()
+
+	router.ServeHTTP(res, req)
+
+	var out response.R
+	if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if out.Code == 0 {
+		t.Fatalf("expected non-zero code when flow exceeded")
+	}
+	if !strings.Contains(out.Msg, "流量") {
+		t.Fatalf("expected flow exceeded message, got %q", out.Msg)
+	}
+
+	status := mustQueryInt(t, repo, `SELECT status FROM forward WHERE id = ?`, forwardID)
+	if status != 0 {
+		t.Fatalf("expected forward status to remain 0, got %d", status)
+	}
+}
+
+func TestForwardResumeBlockedWhenUserTunnelFlowExceeded(t *testing.T) {
+	secret := "contract-jwt-secret"
+	router, repo := setupContractRouter(t, secret)
+	now := time.Now().UnixMilli()
+
+	userID := int64(2)
+	tunnelID := int64(1)
+	forwardID := int64(1)
+
+	if err := repo.DB().Exec(`
+		INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+		VALUES(?, 'ut_flow_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
+	`, userID, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+
+	if err := repo.DB().Exec(`
+		INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, 'ut_flow_tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0)
+	`, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+
+	utFlowGB := int64(120)
+	utUsed := utFlowGB * contractBytesPerGB
+	if err := repo.DB().Exec(`
+		INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+		VALUES(10, ?, ?, NULL, 99999, ?, ?, 0, 1, 2727251700000, 1)
+	`, userID, tunnelID, utFlowGB, utUsed).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+	if err := repo.DB().Exec(`
+		INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
+		VALUES(?, ?, 'ut_flow_user', 'ut_flow_forward', ?, '8.8.8.8:53', 'fifo', 0, 0, ?, ?, 0, 0)
+	`, forwardID, userID, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+
+	token, err := auth.GenerateToken(userID, "ut_flow_user", 1, secret)
+	if err != nil {
+		t.Fatalf("generate token: %v", err)
+	}
+
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/resume", bytes.NewBufferString(`{"id":1}`))
+	req.Header.Set("Authorization", token)
+	req.Header.Set("Content-Type", "application/json")
+	res := httptest.NewRecorder()
+
+	router.ServeHTTP(res, req)
+
+	var out response.R
+	if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if out.Code == 0 {
+		t.Fatalf("expected non-zero code when tunnel flow exceeded")
+	}
+	if !strings.Contains(out.Msg, "隧道") || !strings.Contains(out.Msg, "流量") {
+		t.Fatalf("expected tunnel flow exceeded message, got %q", out.Msg)
+	}
+}
+
+func TestForwardCreateBlockedWhenFlowExceeded(t *testing.T) {
+	secret := "contract-jwt-secret"
+	router, repo := setupContractRouter(t, secret)
+	now := time.Now().UnixMilli()
+
+	userID := int64(2)
+	tunnelID := int64(1)
+
+	flowGB := int64(120)
+	used := flowGB*contractBytesPerGB + 1
+
+	if err := repo.DB().Exec(`
+		INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+		VALUES(?, 'create_flow_user', 'pwd', 1, 2727251700000, ?, ?, 0, 1, 99999, ?, ?, 1)
+	`, userID, flowGB, used, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+
+	if err := repo.DB().Exec(`
+		INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, 'create_flow_tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0)
+	`, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	if err := repo.DB().Exec(`
+		INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+		VALUES(10, ?, ?, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)
+	`, userID, tunnelID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+
+	token, err := auth.GenerateToken(userID, "create_flow_user", 1, secret)
+	if err != nil {
+		t.Fatalf("generate token: %v", err)
+	}
+
+	payload := `{"tunnelId":1,"name":"n","remoteAddr":"1.1.1.1:53"}`
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewBufferString(payload))
+	req.Header.Set("Authorization", token)
+	req.Header.Set("Content-Type", "application/json")
+	res := httptest.NewRecorder()
+
+	router.ServeHTTP(res, req)
+
+	var out response.R
+	if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if out.Code == 0 {
+		t.Fatalf("expected non-zero code when flow exceeded")
+	}
+	if !strings.Contains(out.Msg, "流量") {
+		t.Fatalf("expected flow exceeded message, got %q", out.Msg)
+	}
+}

go-backend/tests/contract/limiter_sync_failure_contract_test.go

@@ -667,6 +667,421 @@ func TestTunnelUpdateRecoversFromAddressInUseContract(t *testing.T) {
 	}
 }
 
+func TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract(t *testing.T) {
+	secret := "contract-jwt-secret"
+	router, r := setupContractRouter(t, secret)
+	server := httptest.NewServer(router)
+	defer server.Close()
+
+	adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
+	if err != nil {
+		t.Fatalf("generate admin token: %v", err)
+	}
+
+	now := time.Now().UnixMilli()
+	if err := r.DB().Exec(`
+		INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+		VALUES(2, 'issue281_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
+	`, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+		INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, "issue281-tunnel", 1.0, 2, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	tunnelID := mustLastInsertID(t, r, "issue281-tunnel")
+
+	insertNode := func(name, secretValue, ip, portRange string, inx int) int64 {
+		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)
+			VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		`, name, secretValue, ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", inx).Error; err != nil {
+			t.Fatalf("insert node %s: %v", name, err)
+		}
+		return mustLastInsertID(t, r, name)
+	}
+
+	oldEntryNodeID := insertNode("issue281-old-entry", "issue281-old-entry-secret", "10.51.0.1", "51000-51010", 0)
+	newEntryNodeID := insertNode("issue281-new-entry", "issue281-new-entry-secret", "10.51.0.2", "52000-52010", 1)
+	exitNodeID := insertNode("issue281-exit", "issue281-exit-secret", "10.51.0.3", "53000-53010", 2)
+
+	if err := r.DB().Exec(`
+		INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
+		VALUES(?, 1, ?, 51001, 'round', 1, 'tls')
+	`, tunnelID, oldEntryNodeID).Error; err != nil {
+		t.Fatalf("insert old entry chain_tunnel: %v", err)
+	}
+	if err := r.DB().Exec(`
+		INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
+		VALUES(?, 3, ?, 53001, 'round', 1, 'tls')
+	`, tunnelID, exitNodeID).Error; err != nil {
+		t.Fatalf("insert exit chain_tunnel: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+		INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+		VALUES(281, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
+	`, tunnelID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+
+	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(2, 'issue281_user', 'issue281-forward', ?, '8.8.8.8:53', 'fifo', 0, 0, ?, ?, 1, 0)
+	`, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+	forwardID := mustLastInsertID(t, r, "issue281-forward")
+
+	if err := r.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, oldEntryNodeID, 51001).Error; err != nil {
+		t.Fatalf("insert forward_port: %v", err)
+	}
+
+	forwardBase := fmt.Sprintf("%d_%d_%d", forwardID, 2, 281)
+
+	var commandMu sync.Mutex
+	oldEntryDeleteNames := make([]string, 0)
+	newEntryUpdateNames := make([]string, 0)
+
+	recordForwardServiceNames := func(data json.RawMessage, list *[]string) {
+		var serviceList []map[string]interface{}
+		if err := json.Unmarshal(data, &serviceList); err == nil {
+			for _, service := range serviceList {
+				name, _ := service["name"].(string)
+				if strings.HasPrefix(strings.TrimSpace(name), forwardBase) {
+					*list = append(*list, name)
+				}
+			}
+			return
+		}
+
+		var payload map[string]interface{}
+		if err := json.Unmarshal(data, &payload); err != nil {
+			return
+		}
+		if rawServices, ok := payload["services"].([]interface{}); ok {
+			for _, raw := range rawServices {
+				name, _ := raw.(string)
+				if strings.HasPrefix(strings.TrimSpace(name), forwardBase) {
+					*list = append(*list, name)
+				}
+			}
+			return
+		}
+	}
+
+	stopOldEntry := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-old-entry-secret", func(cmdType string, data json.RawMessage) (bool, string) {
+		commandMu.Lock()
+		defer commandMu.Unlock()
+		if strings.EqualFold(strings.TrimSpace(cmdType), "DeleteService") {
+			recordForwardServiceNames(data, &oldEntryDeleteNames)
+		}
+		return false, ""
+	})
+	defer stopOldEntry()
+
+	stopNewEntry := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-new-entry-secret", func(cmdType string, data json.RawMessage) (bool, string) {
+		commandMu.Lock()
+		defer commandMu.Unlock()
+		if strings.EqualFold(strings.TrimSpace(cmdType), "UpdateService") || strings.EqualFold(strings.TrimSpace(cmdType), "AddService") {
+			recordForwardServiceNames(data, &newEntryUpdateNames)
+		}
+		return false, ""
+	})
+	defer stopNewEntry()
+
+	stopExit := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-exit-secret", func(cmdType string, data json.RawMessage) (bool, string) {
+		return false, ""
+	})
+	defer stopExit()
+
+	waitNodeStatus(t, r, oldEntryNodeID, 1)
+	waitNodeStatus(t, r, newEntryNodeID, 1)
+	waitNodeStatus(t, r, exitNodeID, 1)
+
+	payload := map[string]interface{}{
+		"id":           tunnelID,
+		"name":         "issue281-tunnel",
+		"type":         2,
+		"flow":         99999,
+		"trafficRatio": 1.0,
+		"status":       1,
+		"inNodeId": []map[string]interface{}{
+			{"nodeId": newEntryNodeID, "protocol": "tls", "strategy": "round"},
+		},
+		"chainNodes": []interface{}{},
+		"outNodeId": []map[string]interface{}{
+			{"nodeId": exitNodeID, "protocol": "tls", "strategy": "round", "port": 53001},
+		},
+	}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload: %v", err)
+	}
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/update", bytes.NewReader(body))
+	req.Header.Set("Authorization", adminToken)
+	req.Header.Set("Content-Type", "application/json")
+	res := httptest.NewRecorder()
+	router.ServeHTTP(res, req)
+	assertCode(t, res, 0)
+
+	nodeAfter, portAfter := mustQueryInt64Int(t, r, `SELECT node_id, port FROM forward_port WHERE forward_id = ? LIMIT 1`, forwardID)
+	if nodeAfter != newEntryNodeID || portAfter != 51001 {
+		t.Fatalf("expected forward_port rebound to node=%d port=51001, got node=%d port=%d", newEntryNodeID, nodeAfter, portAfter)
+	}
+
+	commandMu.Lock()
+	defer commandMu.Unlock()
+	if len(newEntryUpdateNames) == 0 {
+		t.Fatalf("expected new entry node to receive forward runtime sync for %s", forwardBase)
+	}
+	if len(oldEntryDeleteNames) == 0 {
+		t.Fatalf("expected old entry node to receive forward DeleteService cleanup for %s, got none", forwardBase)
+	}
+}
+
+func TestTunnelUpdateEntryTransitionsCleanupForwardRuntimeContract(t *testing.T) {
+	secret := "contract-jwt-secret"
+	router, r := setupContractRouter(t, secret)
+	server := httptest.NewServer(router)
+	defer server.Close()
+
+	adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
+	if err != nil {
+		t.Fatalf("generate admin token: %v", err)
+	}
+
+	now := time.Now().UnixMilli()
+	if err := r.DB().Exec(`
+		INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+		VALUES(2, 'issue281_transition_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
+	`, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+		INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, "issue281-transition-tunnel", 1.0, 2, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	tunnelID := mustLastInsertID(t, r, "issue281-transition-tunnel")
+
+	insertNode := func(name, secretValue, ip, portRange string, inx int) int64 {
+		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)
+			VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		`, name, secretValue, ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", inx).Error; err != nil {
+			t.Fatalf("insert node %s: %v", name, err)
+		}
+		return mustLastInsertID(t, r, name)
+	}
+
+	entryA := insertNode("issue281-transition-entry-a", "issue281-transition-entry-a-secret", "10.52.0.1", "54000-54010", 0)
+	entryB := insertNode("issue281-transition-entry-b", "issue281-transition-entry-b-secret", "10.52.0.2", "55000-55010", 1)
+	entryC := insertNode("issue281-transition-entry-c", "issue281-transition-entry-c-secret", "10.52.0.3", "56000-56010", 2)
+	exitNodeID := insertNode("issue281-transition-exit", "issue281-transition-exit-secret", "10.52.0.4", "57000-57010", 3)
+
+	if err := r.DB().Exec(`
+		INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
+		VALUES(?, 1, ?, 54001, 'round', 1, 'tls')
+	`, tunnelID, entryA).Error; err != nil {
+		t.Fatalf("insert initial entry chain_tunnel: %v", err)
+	}
+	if err := r.DB().Exec(`
+		INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
+		VALUES(?, 3, ?, 57001, 'round', 1, 'tls')
+	`, tunnelID, exitNodeID).Error; err != nil {
+		t.Fatalf("insert exit chain_tunnel: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+		INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+		VALUES(282, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
+	`, tunnelID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+
+	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(2, 'issue281_transition_user', 'issue281-transition-forward', ?, '1.1.1.1:443', 'fifo', 0, 0, ?, ?, 1, 0)
+	`, tunnelID, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+	forwardID := mustLastInsertID(t, r, "issue281-transition-forward")
+
+	if err := r.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, entryA, 54001).Error; err != nil {
+		t.Fatalf("insert forward_port: %v", err)
+	}
+
+	forwardBase := fmt.Sprintf("%d_%d_%d", forwardID, 2, 282)
+	recorder := newForwardRuntimeCommandRecorder(forwardBase)
+
+	stopEntryA := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-entry-a-secret", recorder.handler("entry-a"))
+	defer stopEntryA()
+	stopEntryB := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-entry-b-secret", recorder.handler("entry-b"))
+	defer stopEntryB()
+	stopEntryC := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-entry-c-secret", recorder.handler("entry-c"))
+	defer stopEntryC()
+	stopExit := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-exit-secret", recorder.handler("exit"))
+	defer stopExit()
+
+	waitNodeStatus(t, r, entryA, 1)
+	waitNodeStatus(t, r, entryB, 1)
+	waitNodeStatus(t, r, entryC, 1)
+	waitNodeStatus(t, r, exitNodeID, 1)
+
+	updateTunnelEntries := func(entries []map[string]interface{}) {
+		payload := map[string]interface{}{
+			"id":           tunnelID,
+			"name":         "issue281-transition-tunnel",
+			"type":         2,
+			"flow":         99999,
+			"trafficRatio": 1.0,
+			"status":       1,
+			"inNodeId":     entries,
+			"chainNodes":   []interface{}{},
+			"outNodeId": []map[string]interface{}{
+				{"nodeId": exitNodeID, "protocol": "tls", "strategy": "round", "port": 57001},
+			},
+		}
+		body, err := json.Marshal(payload)
+		if err != nil {
+			t.Fatalf("marshal payload: %v", err)
+		}
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/update", bytes.NewReader(body))
+		req.Header.Set("Authorization", adminToken)
+		req.Header.Set("Content-Type", "application/json")
+		res := httptest.NewRecorder()
+		router.ServeHTTP(res, req)
+		assertCode(t, res, 0)
+	}
+
+	updateTunnelEntries([]map[string]interface{}{
+		{"nodeId": entryA, "protocol": "tls", "strategy": "round"},
+		{"nodeId": entryB, "protocol": "tls", "strategy": "round"},
+	})
+
+	afterMulti := mustQueryNodePorts(t, r, `SELECT node_id, port FROM forward_port WHERE forward_id = ? ORDER BY id ASC`, forwardID)
+	if len(afterMulti) != 2 || afterMulti[entryA] != 54001 || afterMulti[entryB] != 54001 {
+		t.Fatalf("expected forward_port on entryA+entryB with port 54001, got %v", afterMulti)
+	}
+	if recorder.syncCount("entry-b") == 0 {
+		t.Fatalf("expected entry-b to receive forward runtime sync for %s", forwardBase)
+	}
+	if recorder.deleteCount("entry-a") != 0 {
+		t.Fatalf("expected no cleanup on retained entry-a during single->multi transition, got %v", recorder.deleteNames("entry-a"))
+	}
+
+	updateTunnelEntries([]map[string]interface{}{
+		{"nodeId": entryC, "protocol": "tls", "strategy": "round"},
+	})
+
+	afterSingle := mustQueryNodePorts(t, r, `SELECT node_id, port FROM forward_port WHERE forward_id = ? ORDER BY id ASC`, forwardID)
+	if len(afterSingle) != 1 || afterSingle[entryC] != 54001 {
+		t.Fatalf("expected forward_port on entryC with port 54001, got %v", afterSingle)
+	}
+	if recorder.deleteCount("entry-a") == 0 {
+		t.Fatalf("expected cleanup on removed entry-a during multi->single transition, got %v", recorder.deleteNames("entry-a"))
+	}
+	if recorder.deleteCount("entry-b") == 0 {
+		t.Fatalf("expected cleanup on removed entry-b during multi->single transition, got %v", recorder.deleteNames("entry-b"))
+	}
+	if recorder.syncCount("entry-c") == 0 {
+		t.Fatalf("expected entry-c to receive forward runtime sync for %s", forwardBase)
+	}
+}
+
+type forwardRuntimeCommandRecorder struct {
+	prefix string
+
+	mu        sync.Mutex
+	deletes   map[string][]string
+	syncNames map[string][]string
+}
+
+func newForwardRuntimeCommandRecorder(prefix string) *forwardRuntimeCommandRecorder {
+	return &forwardRuntimeCommandRecorder{
+		prefix:    strings.TrimSpace(prefix),
+		deletes:   make(map[string][]string),
+		syncNames: make(map[string][]string),
+	}
+}
+
+func (r *forwardRuntimeCommandRecorder) handler(node string) func(string, json.RawMessage) (bool, string) {
+	return func(cmdType string, data json.RawMessage) (bool, string) {
+		names := collectForwardServiceNames(data, r.prefix)
+		if len(names) == 0 {
+			return false, ""
+		}
+
+		r.mu.Lock()
+		defer r.mu.Unlock()
+
+		if strings.EqualFold(strings.TrimSpace(cmdType), "DeleteService") {
+			r.deletes[node] = append(r.deletes[node], names...)
+		}
+		if strings.EqualFold(strings.TrimSpace(cmdType), "UpdateService") || strings.EqualFold(strings.TrimSpace(cmdType), "AddService") {
+			r.syncNames[node] = append(r.syncNames[node], names...)
+		}
+		return false, ""
+	}
+}
+
+func (r *forwardRuntimeCommandRecorder) deleteCount(node string) int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.deletes[node])
+}
+
+func (r *forwardRuntimeCommandRecorder) syncCount(node string) int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.syncNames[node])
+}
+
+func (r *forwardRuntimeCommandRecorder) deleteNames(node string) []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]string(nil), r.deletes[node]...)
+}
+
+func collectForwardServiceNames(data json.RawMessage, prefix string) []string {
+	prefix = strings.TrimSpace(prefix)
+	if prefix == "" {
+		return nil
+	}
+
+	names := make([]string, 0)
+	var serviceList []map[string]interface{}
+	if err := json.Unmarshal(data, &serviceList); err == nil {
+		for _, service := range serviceList {
+			name, _ := service["name"].(string)
+			if strings.HasPrefix(strings.TrimSpace(name), prefix) {
+				names = append(names, name)
+			}
+		}
+		return names
+	}
+
+	var payload map[string]interface{}
+	if err := json.Unmarshal(data, &payload); err != nil {
+		return nil
+	}
+	if rawServices, ok := payload["services"].([]interface{}); ok {
+		for _, raw := range rawServices {
+			name, _ := raw.(string)
+			if strings.HasPrefix(strings.TrimSpace(name), prefix) {
+				names = append(names, name)
+			}
+		}
+	}
+	return names
+}
+
 func startMockNodeSessionWithCommandFailures(t *testing.T, baseURL string, nodeSecret string, failCommands map[string]string) func() {
 	t.Helper()
 

plans/028-tunnel-entry-change-sync-forward-ports.md

@@ -0,0 +1,19 @@
+# 028 - Sync Forward Ports On Tunnel Entry Change
+
+## Goal
+When a tunnel's entry nodes change, automatically keep all forwards under that tunnel aligned by rebuilding `forward_port` rows to match the latest entry node set.
+
+## Scope
+- Backend only: update tunnel mutation flow to sync forward entry mappings.
+- Preserve existing forward port and bind IP behavior:
+  - Keep the existing forward port (choose the current min port in `forward_port`).
+  - Preserve `in_ip` only when the tunnel has a single entry node; clear `in_ip` for multi-entry tunnels.
+
+## Checklist
+- [x] Capture old entry node IDs before tunnel update commits.
+- [x] After commit, compare old/new entry node sets.
+- [x] If changed, rebuild `forward_port` for all forwards in the tunnel.
+- [x] Run `go test ./...` in `go-backend`.
+
+## Notes
+- Runtime redeploy/downlink is handled elsewhere; this change focuses on DB-level consistency of forward entry mappings.

plans/029-issue-281-contract-repro.md

@@ -0,0 +1,14 @@
+# 029 - Issue 281 Contract Repro
+
+## Goal
+Add a contract test that reproduces issue #281: after changing a tunnel's entry node, forward runtime cleanup does not remove the stale service from the old entry node.
+
+## Checklist
+- [x] Review existing contract test helpers for mock node command recording.
+- [x] Add a contract test that updates a tunnel entry node while a forward is bound to the tunnel.
+- [x] Assert the new entry node receives forward sync commands and the old entry node does not receive forward cleanup, reproducing the bug.
+- [x] Run the focused contract test and capture the failure.
+
+## Test Record
+- Command: `cd go-backend && go test ./tests/contract/... -run TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract`
+- Result: failed as expected with `expected old entry node to receive forward DeleteService cleanup for 1_2_281, got none`.

plans/030-fix-issue-281-stale-forward-runtime-cleanup.md

@@ -0,0 +1,14 @@
+# 030 - Fix Issue 281 Stale Forward Runtime Cleanup
+
+## Goal
+When a tunnel's entry nodes change, remove forward runtime services from entry nodes that are no longer part of the tunnel before syncing the forward to its new entry nodes.
+
+## Checklist
+- [x] Review the tunnel update flow and identify where old/new entry node sets are available.
+- [x] Add backend cleanup for forward runtimes on removed entry nodes.
+- [x] Keep existing forward port rebuild and forward resync behavior intact.
+- [x] Run focused contract regression tests for the issue 281 repro.
+
+## Test Record
+- Command: `cd go-backend && go test ./tests/contract/... -run 'TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract|TestTunnelUpdateRecoversFromAddressInUseContract'`
+- Result: passed.

plans/031-entry-transition-regression-coverage.md

@@ -0,0 +1,14 @@
+# 031 - Entry Transition Regression Coverage
+
+## Goal
+Expand issue #281 regression coverage to verify forward runtime cleanup and `forward_port` rebuilding across both single-entry to multi-entry and multi-entry to single-entry tunnel updates.
+
+## Checklist
+- [x] Review the current issue 281 contract repro and reuse its mock-node recording helpers.
+- [x] Add a broader contract test that exercises both entry transition directions.
+- [x] Assert removed entry nodes receive forward cleanup and retained/new entry nodes receive forward sync.
+- [x] Run focused contract tests and record the result.
+
+## Test Record
+- Command: `cd go-backend && go test ./tests/contract/... -run 'TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract|TestTunnelUpdateEntryTransitionsCleanupForwardRuntimeContract|TestTunnelUpdateRecoversFromAddressInUseContract'`
+- Result: passed.

vite-frontend/src/pages/node.tsx

@@ -1779,7 +1779,7 @@ export default function NodePage() {
                                   )}
                                 </button>
                                 <div
-                                  className={`pointer-events-none invisible absolute z-30 w-72 max-w-[min(18rem,calc(100vw-4rem))] rounded-xl border border-divider/80 bg-background/98 p-3 opacity-0 shadow-xl backdrop-blur transition-all duration-150 group-hover/info:visible group-hover/info:pointer-events-auto group-hover/info:opacity-100 group-focus-within/info:visible group-focus-within/info:pointer-events-auto group-focus-within/info:opacity-100 ${
+                                  className={`pointer-events-none invisible absolute z-[60] w-72 max-w-[min(18rem,calc(100vw-4rem))] rounded-xl border border-divider/80 bg-background/98 p-3 opacity-0 shadow-xl backdrop-blur transition-all duration-150 group-hover/info:visible group-hover/info:pointer-events-auto group-hover/info:opacity-100 group-focus-within/info:visible group-focus-within/info:pointer-events-auto group-focus-within/info:opacity-100 ${
                                     infoPlacement === "bottom"
                                       ? "right-0 top-[calc(100%+0.75rem)] translate-y-1 group-hover/info:translate-y-0 group-focus-within/info:translate-y-0"
                                       : "right-[calc(100%+0.75rem)] top-1/2 -translate-y-1/2 translate-x-1 group-hover/info:translate-x-0 group-focus-within/info:translate-x-0"