mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-09-28 07:36:38 +08:00
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
This commit is contained in:
@@ -286,6 +286,9 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all
|
|||||||
if err != nil && allowFallbackAdd && method == "UpdateService" {
|
if err != nil && allowFallbackAdd && method == "UpdateService" {
|
||||||
_, err = h.sendNodeCommand(node.ID, "AddService", services, true, false)
|
_, 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 {
|
if err != nil {
|
||||||
return fmt.Errorf("节点 %s 下发失败: %w", node.Name, err)
|
return fmt.Errorf("节点 %s 下发失败: %w", node.Name, err)
|
||||||
}
|
}
|
||||||
@@ -293,6 +296,77 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all
|
|||||||
return nil
|
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 {
|
func (h *Handler) controlForwardServices(forward *forwardRecord, commandType string, tolerateNotFound bool) error {
|
||||||
if h == nil || forward == nil {
|
if h == nil || forward == nil {
|
||||||
return errors.New("invalid forward control context")
|
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, "已存在")
|
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{} {
|
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"}
|
protocols := []string{"tcp", "udp"}
|
||||||
services := make([]map[string]interface{}, 0, 2)
|
services := make([]map[string]interface{}, 0, 2)
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package handler
|
package handler
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"reflect"
|
"reflect"
|
||||||
"testing"
|
"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) {
|
func TestBuildForwardServiceConfigs_UsesBindIPForListen(t *testing.T) {
|
||||||
forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7}
|
forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7}
|
||||||
node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"}
|
node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"}
|
||||||
|
|||||||
@@ -111,6 +111,25 @@ func (r *Repository) ListForwardPorts(forwardID int64) ([]model.ForwardPortRecor
|
|||||||
return rows, nil
|
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) {
|
func (r *Repository) GetTunnelOutProtocol(tunnelID int64) (string, error) {
|
||||||
if r == nil || r.db == nil {
|
if r == nil || r.db == nil {
|
||||||
return "", errors.New("repository not initialized")
|
return "", errors.New("repository not initialized")
|
||||||
|
|||||||
@@ -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.
|
||||||
Reference in New Issue
Block a user