From 7efb49bdab04a0163f5ac7e7754988f8f684a67d Mon Sep 17 00:00:00 2001 From: sagitchu Date: Tue, 3 Mar 2026 20:53:55 +0800 Subject: [PATCH] fix: handle forward service rebind on self-occupied port When UpdateService encounters bind address conflicts (port already in use), the handler now automatically deletes existing forward services and retries the AddService operation. This resolves issues where a forward's own stale listener prevents the update. - Add isBindAddressInUseError() to detect port bind conflicts - Add rebindForwardServiceOnSelfOccupiedPort() for automatic cleanup and retry - Add HasOtherForwardOnNodePort() repository method to verify port ownership - Add unit tests for bind conflict detection --- .../internal/http/handler/control_plane.go | 88 +++++++++++++++++++ .../http/handler/control_plane_test.go | 16 ++++ .../internal/store/repo/repository_control.go | 19 ++++ ...rward-explicit-bind-self-occupy-release.md | 11 +++ 4 files changed, 134 insertions(+) create mode 100644 plans/004-forward-explicit-bind-self-occupy-release.md diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 0dfce59..b4ed183 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -286,6 +286,9 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all if err != nil && allowFallbackAdd && method == "UpdateService" { _, err = h.sendNodeCommand(node.ID, "AddService", services, true, false) } + if err != nil && strings.EqualFold(strings.TrimSpace(method), "UpdateService") && isBindAddressInUseError(err) { + err = h.rebindForwardServiceOnSelfOccupiedPort(forward, node, fp.Port, services) + } if err != nil { return fmt.Errorf("节点 %s 下发失败: %w", node.Name, err) } @@ -293,6 +296,77 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all return nil } +func (h *Handler) rebindForwardServiceOnSelfOccupiedPort(forward *forwardRecord, node *nodeRecord, port int, services []map[string]interface{}) error { + if h == nil || forward == nil || node == nil { + return errors.New("invalid self-occupy rebind context") + } + if port <= 0 { + return errors.New("invalid forward port") + } + + hasOtherForward, err := h.repo.HasOtherForwardOnNodePort(node.ID, port, forward.ID) + if err != nil { + return err + } + if hasOtherForward { + return fmt.Errorf("端口 %d 已被其他转发占用", port) + } + + if err := h.deleteForwardServicesOnNode(forward, node.ID); err != nil { + return err + } + + time.Sleep(150 * time.Millisecond) + + _, err = h.sendNodeCommand(node.ID, "AddService", services, true, false) + if err != nil { + return err + } + + return nil +} + +func (h *Handler) deleteForwardServicesOnNode(forward *forwardRecord, nodeID int64) error { + if h == nil || forward == nil { + return errors.New("invalid forward delete context") + } + + userTunnelID, _, _, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID) + if err != nil { + return err + } + userTunnelIDs, err := h.listUserTunnelIDs(forward.UserID, forward.TunnelID) + if err != nil { + return err + } + allUserTunnelIDs, err := h.listUserTunnelIDsByUser(forward.UserID) + if err != nil { + return err + } + candidateTunnelIDs := make([]int64, 0, len(userTunnelIDs)+len(allUserTunnelIDs)) + candidateTunnelIDs = append(candidateTunnelIDs, userTunnelIDs...) + candidateTunnelIDs = append(candidateTunnelIDs, allUserTunnelIDs...) + bases := buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, candidateTunnelIDs) + + var lastErr error + for _, base := range bases { + names := buildForwardControlServiceNames(base, "DeleteService") + payload := map[string]interface{}{ + "services": names, + } + _, cmdErr := h.sendNodeCommand(nodeID, "DeleteService", payload, false, true) + if cmdErr == nil { + return nil + } + lastErr = cmdErr + } + + if lastErr != nil { + return lastErr + } + return nil +} + func (h *Handler) controlForwardServices(forward *forwardRecord, commandType string, tolerateNotFound bool) error { if h == nil || forward == nil { return errors.New("invalid forward control context") @@ -1303,6 +1377,20 @@ func isAlreadyExistsMessage(message string) bool { return strings.Contains(msg, "already exists") || strings.Contains(msg, "已存在") } +func isBindAddressInUseError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(strings.TrimSpace(err.Error())) + if msg == "" { + return false + } + if strings.Contains(msg, "address already in use") { + return true + } + return strings.Contains(msg, "cannot assign requested address") +} + func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, limiterID *int64, tunnelTLSProtocol bool) []map[string]interface{} { protocols := []string{"tcp", "udp"} services := make([]map[string]interface{}, 0, 2) diff --git a/go-backend/internal/http/handler/control_plane_test.go b/go-backend/internal/http/handler/control_plane_test.go index 1919f0b..02d8586 100644 --- a/go-backend/internal/http/handler/control_plane_test.go +++ b/go-backend/internal/http/handler/control_plane_test.go @@ -1,6 +1,7 @@ package handler import ( + "errors" "reflect" "testing" ) @@ -66,6 +67,21 @@ func TestIsAlreadyExistsMessage(t *testing.T) { } } +func TestIsBindAddressInUseError(t *testing.T) { + if !isBindAddressInUseError(errors.New("listen tcp [::]:10001: bind: address already in use")) { + t.Fatalf("address already in use should be detected") + } + if !isBindAddressInUseError(errors.New("listen tcp4 13.228.170.187:16765: bind: cannot assign requested address")) { + t.Fatalf("cannot assign requested address should be detected") + } + if isBindAddressInUseError(errors.New("service demo already exists")) { + t.Fatalf("already exists should not be treated as bind conflict") + } + if isBindAddressInUseError(nil) { + t.Fatalf("nil error should not be treated as bind conflict") + } +} + func TestBuildForwardServiceConfigs_UsesBindIPForListen(t *testing.T) { forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"} diff --git a/go-backend/internal/store/repo/repository_control.go b/go-backend/internal/store/repo/repository_control.go index 2799486..7bc0922 100644 --- a/go-backend/internal/store/repo/repository_control.go +++ b/go-backend/internal/store/repo/repository_control.go @@ -111,6 +111,25 @@ func (r *Repository) ListForwardPorts(forwardID int64) ([]model.ForwardPortRecor return rows, nil } +func (r *Repository) HasOtherForwardOnNodePort(nodeID int64, port int, currentForwardID int64) (bool, error) { + if r == nil || r.db == nil { + return false, errors.New("repository not initialized") + } + if nodeID <= 0 || port <= 0 { + return false, nil + } + + var count int64 + err := r.db.Model(&model.ForwardPort{}). + Where("node_id = ? AND port = ? AND forward_id <> ?", nodeID, port, currentForwardID). + Count(&count).Error + if err != nil { + return false, err + } + + return count > 0, nil +} + func (r *Repository) GetTunnelOutProtocol(tunnelID int64) (string, error) { if r == nil || r.db == nil { return "", errors.New("repository not initialized") diff --git a/plans/004-forward-explicit-bind-self-occupy-release.md b/plans/004-forward-explicit-bind-self-occupy-release.md new file mode 100644 index 0000000..26dbe5f --- /dev/null +++ b/plans/004-forward-explicit-bind-self-occupy-release.md @@ -0,0 +1,11 @@ +# 004 Forward Explicit Bind Self-Occupy Release + +## Checklist + +- [x] Confirm current forward edit/save failure path and lock strategy: explicit bind always stays explicit. +- [x] Add repository query to detect whether a node+port is occupied by other forwards (excluding current forward). +- [x] Enhance forward service sync to treat address-in-use as a recoverable case when only self occupies the port. +- [x] On self-occupy conflict, proactively delete current forward services on target node and retry AddService. +- [x] Keep hard failure when the same node+port is occupied by other forwards. +- [x] Add focused unit tests for new error classification helpers. +- [x] Run focused backend tests for touched handler/repo packages.