mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-10-07 02:06:38 +08:00
fix: add retry mechanism for tunnel service bind conflicts (#286)
## Summary - Add retry logic for tunnel service creation/update when encountering "address already in use" bind errors - Automatically cleanup stale service and retry once before failing - Add comprehensive unit and contract tests for bind conflict scenarios ## Changes - `mutations.go`: Add `retryTunnelServiceAddWithCleanup` helper and `addTunnelServiceOnNode` wrapper - `control_plane_test.go`: Unit tests for retry behavior on address conflicts - `dual_stack_test.go`: Test fallback to node listen address - `forward_contract_test.go`: Contract test for forward update with bind retry - `limiter_sync_failure_contract_test.go`: Contract test for tunnel update with bind retry - `plans/016-tunnel-runtime-bind-conflict-retry.md`: Implementation plan document ## Test Plan - Unit tests verify retry logic executes correctly - Contract tests validate end-to-end behavior with mock nodes - All tests pass with race detector enabled
This commit is contained in:
@@ -273,6 +273,75 @@ func TestIsCannotAssignRequestedAddressError(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRetryTunnelServiceAddWithCleanupRetriesOnAddressInUse(t *testing.T) {
|
||||||
|
addCalls := 0
|
||||||
|
cleanupCalls := 0
|
||||||
|
err := retryTunnelServiceAddWithCleanup(
|
||||||
|
func() error {
|
||||||
|
addCalls++
|
||||||
|
if addCalls == 1 {
|
||||||
|
return errors.New("listen tcp 10.0.0.1:32000: bind: address already in use")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
func() error {
|
||||||
|
cleanupCalls++
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
0,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("expected retry to succeed, got %v", err)
|
||||||
|
}
|
||||||
|
if addCalls != 2 {
|
||||||
|
t.Fatalf("expected 2 add attempts, got %d", addCalls)
|
||||||
|
}
|
||||||
|
if cleanupCalls != 1 {
|
||||||
|
t.Fatalf("expected 1 cleanup attempt, got %d", cleanupCalls)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRetryTunnelServiceAddWithCleanupSkipsCleanupOnNonBindError(t *testing.T) {
|
||||||
|
addCalls := 0
|
||||||
|
cleanupCalls := 0
|
||||||
|
err := retryTunnelServiceAddWithCleanup(
|
||||||
|
func() error {
|
||||||
|
addCalls++
|
||||||
|
return errors.New("network timeout")
|
||||||
|
},
|
||||||
|
func() error {
|
||||||
|
cleanupCalls++
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
0,
|
||||||
|
)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("expected hard error")
|
||||||
|
}
|
||||||
|
if addCalls != 1 {
|
||||||
|
t.Fatalf("expected 1 add attempt, got %d", addCalls)
|
||||||
|
}
|
||||||
|
if cleanupCalls != 0 {
|
||||||
|
t.Fatalf("expected 0 cleanup attempts, got %d", cleanupCalls)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRetryTunnelServiceAddWithCleanupReturnsCleanupError(t *testing.T) {
|
||||||
|
cleanupErr := errors.New("delete failed")
|
||||||
|
err := retryTunnelServiceAddWithCleanup(
|
||||||
|
func() error {
|
||||||
|
return errors.New("listen tcp 10.0.0.1:32000: bind: address already in use")
|
||||||
|
},
|
||||||
|
func() error {
|
||||||
|
return cleanupErr
|
||||||
|
},
|
||||||
|
0,
|
||||||
|
)
|
||||||
|
if !errors.Is(err, cleanupErr) {
|
||||||
|
t.Fatalf("expected cleanup error %v, got %v", cleanupErr, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
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: "[::]"}
|
||||||
|
|||||||
@@ -43,6 +43,19 @@ func TestBuildTunnelChainServiceConfig_UsesConnectIPForListen(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestBuildTunnelChainServiceConfig_FallsBackToNodeListenAddr(t *testing.T) {
|
||||||
|
node := &nodeRecord{TCPListenAddr: "10.8.0.5"}
|
||||||
|
chain := tunnelRuntimeNode{Protocol: "tls", Port: 21002}
|
||||||
|
services := buildTunnelChainServiceConfig(99, chain, node)
|
||||||
|
if len(services) != 1 {
|
||||||
|
t.Fatalf("expected 1 service, got %d", len(services))
|
||||||
|
}
|
||||||
|
addr, _ := services[0]["addr"].(string)
|
||||||
|
if addr != "10.8.0.5:21002" {
|
||||||
|
t.Fatalf("expected node listen addr 10.8.0.5:21002, got %q", addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestBuildTunnelChainServiceConfig_DefaultListenAddrWhenConnectIPEmpty(t *testing.T) {
|
func TestBuildTunnelChainServiceConfig_DefaultListenAddrWhenConnectIPEmpty(t *testing.T) {
|
||||||
node := &nodeRecord{TCPListenAddr: "[::]"}
|
node := &nodeRecord{TCPListenAddr: "[::]"}
|
||||||
chain := tunnelRuntimeNode{Protocol: "tls", Port: 21001}
|
chain := tunnelRuntimeNode{Protocol: "tls", Port: 21001}
|
||||||
|
|||||||
@@ -25,6 +25,8 @@ import (
|
|||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const tunnelServiceBindRetryDelay = 150 * time.Millisecond
|
||||||
|
|
||||||
func (h *Handler) userCreate(w http.ResponseWriter, r *http.Request) {
|
func (h *Handler) userCreate(w http.ResponseWriter, r *http.Request) {
|
||||||
if r.Method != http.MethodPost {
|
if r.Method != http.MethodPost {
|
||||||
response.WriteJSON(w, response.ErrDefault("请求失败"))
|
response.WriteJSON(w, response.ErrDefault("请求失败"))
|
||||||
@@ -2584,7 +2586,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64
|
|||||||
createdChains = append(createdChains, chainNode.NodeID)
|
createdChains = append(createdChains, chainNode.NodeID)
|
||||||
|
|
||||||
serviceData := buildTunnelChainServiceConfig(state.TunnelID, chainNode, state.Nodes[chainNode.NodeID])
|
serviceData := buildTunnelChainServiceConfig(state.TunnelID, chainNode, state.Nodes[chainNode.NodeID])
|
||||||
if _, err := h.sendNodeCommand(chainNode.NodeID, "AddService", serviceData, true, false); err != nil {
|
if err := h.addTunnelServiceOnNode(chainNode.NodeID, state.TunnelID, serviceData); err != nil {
|
||||||
return createdChains, createdServices, fmt.Errorf("转发链节点 %s 下发服务失败: %w", nodeDisplayName(state.Nodes[chainNode.NodeID]), err)
|
return createdChains, createdServices, fmt.Errorf("转发链节点 %s 下发服务失败: %w", nodeDisplayName(state.Nodes[chainNode.NodeID]), err)
|
||||||
}
|
}
|
||||||
createdServices = append(createdServices, chainNode.NodeID)
|
createdServices = append(createdServices, chainNode.NodeID)
|
||||||
@@ -2596,7 +2598,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
serviceData := buildTunnelChainServiceConfig(state.TunnelID, outNode, state.Nodes[outNode.NodeID])
|
serviceData := buildTunnelChainServiceConfig(state.TunnelID, outNode, state.Nodes[outNode.NodeID])
|
||||||
if _, err := h.sendNodeCommand(outNode.NodeID, "AddService", serviceData, true, false); err != nil {
|
if err := h.addTunnelServiceOnNode(outNode.NodeID, state.TunnelID, serviceData); err != nil {
|
||||||
return createdChains, createdServices, fmt.Errorf("出口节点 %s 下发服务失败: %w", nodeDisplayName(state.Nodes[outNode.NodeID]), err)
|
return createdChains, createdServices, fmt.Errorf("出口节点 %s 下发服务失败: %w", nodeDisplayName(state.Nodes[outNode.NodeID]), err)
|
||||||
}
|
}
|
||||||
createdServices = append(createdServices, outNode.NodeID)
|
createdServices = append(createdServices, outNode.NodeID)
|
||||||
@@ -2605,6 +2607,44 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64
|
|||||||
return createdChains, createdServices, nil
|
return createdChains, createdServices, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func retryTunnelServiceAddWithCleanup(add func() error, cleanup func() error, wait time.Duration) error {
|
||||||
|
if add == nil {
|
||||||
|
return errors.New("invalid tunnel service add callback")
|
||||||
|
}
|
||||||
|
err := add()
|
||||||
|
if err == nil || !isAddressAlreadyInUseError(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if cleanup == nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if cleanupErr := cleanup(); cleanupErr != nil {
|
||||||
|
return cleanupErr
|
||||||
|
}
|
||||||
|
if wait > 0 {
|
||||||
|
time.Sleep(wait)
|
||||||
|
}
|
||||||
|
return add()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *Handler) addTunnelServiceOnNode(nodeID, tunnelID int64, serviceData []map[string]interface{}) error {
|
||||||
|
if h == nil {
|
||||||
|
return errors.New("invalid tunnel service context")
|
||||||
|
}
|
||||||
|
serviceName := fmt.Sprintf("%d_tls", tunnelID)
|
||||||
|
return retryTunnelServiceAddWithCleanup(
|
||||||
|
func() error {
|
||||||
|
_, err := h.sendNodeCommand(nodeID, "AddService", serviceData, true, false)
|
||||||
|
return err
|
||||||
|
},
|
||||||
|
func() error {
|
||||||
|
_, err := h.sendNodeCommand(nodeID, "DeleteService", map[string]interface{}{"services": []string{serviceName}}, false, true)
|
||||||
|
return err
|
||||||
|
},
|
||||||
|
tunnelServiceBindRetryDelay,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
func (h *Handler) rollbackTunnelRuntime(chainNodeIDs, serviceNodeIDs []int64, tunnelID int64) {
|
func (h *Handler) rollbackTunnelRuntime(chainNodeIDs, serviceNodeIDs []int64, tunnelID int64) {
|
||||||
if h == nil || tunnelID <= 0 {
|
if h == nil || tunnelID <= 0 {
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -914,6 +916,164 @@ func TestForwardCreateThenPauseResumeContract(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestForwardUpdateRecoversFromAddressInUseContract(t *testing.T) {
|
||||||
|
secret := "contract-jwt-secret"
|
||||||
|
router, repo := 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 := 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(202, 'forward_bind_retry_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
|
||||||
|
`, now, now).Error; err != nil {
|
||||||
|
t.Fatalf("insert user: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := repo.DB().Exec(`
|
||||||
|
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
|
||||||
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "forward-bind-retry-tunnel", 1.0, 1, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
|
||||||
|
t.Fatalf("insert tunnel: %v", err)
|
||||||
|
}
|
||||||
|
tunnelID := mustLastInsertID(t, repo, "forward-bind-retry-tunnel")
|
||||||
|
|
||||||
|
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)
|
||||||
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "forward-bind-retry-node", "forward-bind-retry-secret", "10.42.0.1", "10.42.0.1", "", "44000-44010", "", "v1", 1, 1, 1, now, now, 1, "10.42.0.9", "[::]", 0).Error; err != nil {
|
||||||
|
t.Fatalf("insert node: %v", err)
|
||||||
|
}
|
||||||
|
nodeID := mustLastInsertID(t, repo, "forward-bind-retry-node")
|
||||||
|
|
||||||
|
if err := repo.DB().Exec(`
|
||||||
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
||||||
|
VALUES(?, 1, ?, 44001, 'round', 1, 'tls')
|
||||||
|
`, tunnelID, nodeID).Error; err != nil {
|
||||||
|
t.Fatalf("insert chain_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(41, 202, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
|
||||||
|
`, tunnelID).Error; err != nil {
|
||||||
|
t.Fatalf("insert user_tunnel: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
createPayload := map[string]interface{}{
|
||||||
|
"name": "forward-bind-retry-target",
|
||||||
|
"tunnelId": tunnelID,
|
||||||
|
"remoteAddr": "1.1.1.1:443",
|
||||||
|
"strategy": "fifo",
|
||||||
|
}
|
||||||
|
createBody, err := json.Marshal(createPayload)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal create payload: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var mu sync.Mutex
|
||||||
|
counts := map[string]int{}
|
||||||
|
var addServiceAddrs []string
|
||||||
|
triggerConflict := false
|
||||||
|
stopNode := startMockNodeSessionWithCommandRecorder(t, server.URL, "forward-bind-retry-secret", func(cmdType string, data json.RawMessage) (bool, string) {
|
||||||
|
key := strings.ToLower(strings.TrimSpace(cmdType))
|
||||||
|
mu.Lock()
|
||||||
|
counts[key]++
|
||||||
|
attempt := counts[key]
|
||||||
|
if strings.EqualFold(strings.TrimSpace(cmdType), "AddService") || strings.EqualFold(strings.TrimSpace(cmdType), "UpdateService") {
|
||||||
|
var services []map[string]interface{}
|
||||||
|
if err := json.Unmarshal(data, &services); err == nil {
|
||||||
|
for _, svc := range services {
|
||||||
|
if addr, _ := svc["addr"].(string); strings.TrimSpace(addr) != "" {
|
||||||
|
addServiceAddrs = append(addServiceAddrs, addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
shouldFail := false
|
||||||
|
if triggerConflict {
|
||||||
|
if strings.EqualFold(strings.TrimSpace(cmdType), "UpdateService") && attempt == 1 {
|
||||||
|
shouldFail = true
|
||||||
|
}
|
||||||
|
if strings.EqualFold(strings.TrimSpace(cmdType), "AddService") && attempt == 1 {
|
||||||
|
shouldFail = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
mu.Unlock()
|
||||||
|
if shouldFail {
|
||||||
|
return true, "listen tcp 10.42.0.9:44001: bind: address already in use"
|
||||||
|
}
|
||||||
|
return false, ""
|
||||||
|
})
|
||||||
|
defer stopNode()
|
||||||
|
waitNodeStatus(t, repo, nodeID, 1)
|
||||||
|
|
||||||
|
createReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(createBody))
|
||||||
|
createReq.Header.Set("Authorization", adminToken)
|
||||||
|
createReq.Header.Set("Content-Type", "application/json")
|
||||||
|
createRes := httptest.NewRecorder()
|
||||||
|
router.ServeHTTP(createRes, createReq)
|
||||||
|
assertCode(t, createRes, 0)
|
||||||
|
mu.Lock()
|
||||||
|
counts = map[string]int{}
|
||||||
|
addServiceAddrs = nil
|
||||||
|
triggerConflict = true
|
||||||
|
mu.Unlock()
|
||||||
|
|
||||||
|
forwardID := mustLastInsertID(t, repo, "forward-bind-retry-target")
|
||||||
|
|
||||||
|
updatePayload := map[string]interface{}{
|
||||||
|
"id": forwardID,
|
||||||
|
"name": "forward-bind-retry-target-updated",
|
||||||
|
"tunnelId": tunnelID,
|
||||||
|
"remoteAddr": "9.9.9.9:8443",
|
||||||
|
"strategy": "fifo",
|
||||||
|
}
|
||||||
|
updateBody, err := json.Marshal(updatePayload)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal update payload: %v", err)
|
||||||
|
}
|
||||||
|
updateReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/update", bytes.NewReader(updateBody))
|
||||||
|
updateReq.Header.Set("Authorization", adminToken)
|
||||||
|
updateReq.Header.Set("Content-Type", "application/json")
|
||||||
|
updateRes := httptest.NewRecorder()
|
||||||
|
router.ServeHTTP(updateRes, updateReq)
|
||||||
|
assertCode(t, updateRes, 0)
|
||||||
|
|
||||||
|
mu.Lock()
|
||||||
|
defer mu.Unlock()
|
||||||
|
boundPort := mustQueryInt(t, repo, `SELECT port FROM forward_port WHERE forward_id = ? LIMIT 1`, forwardID)
|
||||||
|
if counts["updateservice"] != 1 {
|
||||||
|
t.Fatalf("expected one UpdateService attempt, got %d (%v)", counts["updateservice"], counts)
|
||||||
|
}
|
||||||
|
if counts["deleteservice"] == 0 {
|
||||||
|
t.Fatalf("expected DeleteService cleanup after address-in-use (%v)", counts)
|
||||||
|
}
|
||||||
|
if counts["addservice"] < 2 {
|
||||||
|
t.Fatalf("expected AddService retry path to run at least twice total, got %d (%v)", counts["addservice"], counts)
|
||||||
|
}
|
||||||
|
foundBindAddr := false
|
||||||
|
for _, addr := range addServiceAddrs {
|
||||||
|
if addr == "10.42.0.9:"+strconv.Itoa(boundPort) {
|
||||||
|
foundBindAddr = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !foundBindAddr {
|
||||||
|
t.Fatalf("expected forward runtime to keep node listen addr 10.42.0.9:%d, got %v", boundPort, addServiceAddrs)
|
||||||
|
}
|
||||||
|
|
||||||
|
storedRemoteAddr := mustQueryString(t, repo, `SELECT remote_addr FROM forward WHERE id = ?`, forwardID)
|
||||||
|
if storedRemoteAddr != "9.9.9.9:8443" {
|
||||||
|
t.Fatalf("expected remote_addr update to persist, got %q", storedRemoteAddr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func jsonNumber(v int64) string {
|
func jsonNumber(v int64) string {
|
||||||
return strconv.FormatInt(v, 10)
|
return strconv.FormatInt(v, 10)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -538,6 +539,134 @@ func TestBatchAssignInsertRollbackWhenLimiterDispatchFailsContract(t *testing.T)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTunnelUpdateRecoversFromAddressInUseContract(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 tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
|
||||||
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "tunnel-bind-retry", 1.0, 1, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
|
||||||
|
t.Fatalf("insert tunnel: %v", err)
|
||||||
|
}
|
||||||
|
tunnelID := mustLastInsertID(t, r, "tunnel-bind-retry")
|
||||||
|
|
||||||
|
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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "tunnel-bind-entry", "tunnel-bind-entry-secret", "10.41.0.1", "10.41.0.1", "", "43000-43010", "", "v1", 1, 1, 1, now, now, 1, "10.41.0.1", "[::]", 0).Error; err != nil {
|
||||||
|
t.Fatalf("insert entry node: %v", err)
|
||||||
|
}
|
||||||
|
entryNodeID := mustLastInsertID(t, r, "tunnel-bind-entry")
|
||||||
|
|
||||||
|
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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "tunnel-bind-exit", "tunnel-bind-exit-secret", "10.41.0.2", "10.41.0.2", "", "43100-43110", "eth0", "v1", 1, 1, 1, now, now, 1, "10.41.0.9", "[::]", 0).Error; err != nil {
|
||||||
|
t.Fatalf("insert exit node: %v", err)
|
||||||
|
}
|
||||||
|
exitNodeID := mustLastInsertID(t, r, "tunnel-bind-exit")
|
||||||
|
|
||||||
|
if err := r.DB().Exec(`
|
||||||
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
||||||
|
VALUES(?, 1, ?, 43001, 'round', 1, 'tls')
|
||||||
|
`, tunnelID, entryNodeID).Error; err != nil {
|
||||||
|
t.Fatalf("insert entry chain_tunnel: %v", err)
|
||||||
|
}
|
||||||
|
if err := r.DB().Exec(`
|
||||||
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol, connect_ip)
|
||||||
|
VALUES(?, 3, ?, 43101, 'round', 1, 'tls', ?)
|
||||||
|
`, tunnelID, exitNodeID, "10.41.0.99").Error; err != nil {
|
||||||
|
t.Fatalf("insert exit chain_tunnel: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var commandMu sync.Mutex
|
||||||
|
commandCounts := map[string]int{}
|
||||||
|
var addServiceAddrs []string
|
||||||
|
stopEntry := startMockNodeSessionWithCommandRecorder(t, server.URL, "tunnel-bind-entry-secret", func(cmdType string, data json.RawMessage) (bool, string) {
|
||||||
|
commandMu.Lock()
|
||||||
|
defer commandMu.Unlock()
|
||||||
|
commandCounts["entry:"+strings.ToLower(strings.TrimSpace(cmdType))]++
|
||||||
|
return false, ""
|
||||||
|
})
|
||||||
|
defer stopEntry()
|
||||||
|
stopExit := startMockNodeSessionWithCommandRecorder(t, server.URL, "tunnel-bind-exit-secret", func(cmdType string, data json.RawMessage) (bool, string) {
|
||||||
|
key := "exit:" + strings.ToLower(strings.TrimSpace(cmdType))
|
||||||
|
commandMu.Lock()
|
||||||
|
commandCounts[key]++
|
||||||
|
attempt := commandCounts[key]
|
||||||
|
if strings.EqualFold(strings.TrimSpace(cmdType), "AddService") {
|
||||||
|
var services []map[string]interface{}
|
||||||
|
if err := json.Unmarshal(data, &services); err == nil {
|
||||||
|
for _, svc := range services {
|
||||||
|
if addr, _ := svc["addr"].(string); strings.TrimSpace(addr) != "" {
|
||||||
|
addServiceAddrs = append(addServiceAddrs, addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
commandMu.Unlock()
|
||||||
|
if strings.EqualFold(strings.TrimSpace(cmdType), "AddService") && attempt == 1 {
|
||||||
|
return true, "listen tcp 10.41.0.99:43101: bind: address already in use"
|
||||||
|
}
|
||||||
|
return false, ""
|
||||||
|
})
|
||||||
|
defer stopExit()
|
||||||
|
waitNodeStatus(t, r, entryNodeID, 1)
|
||||||
|
waitNodeStatus(t, r, exitNodeID, 1)
|
||||||
|
|
||||||
|
payload := map[string]interface{}{
|
||||||
|
"id": tunnelID,
|
||||||
|
"name": "tunnel-bind-retry",
|
||||||
|
"type": 2,
|
||||||
|
"flow": 99999,
|
||||||
|
"trafficRatio": 1.0,
|
||||||
|
"status": 1,
|
||||||
|
"inNodeId": []map[string]interface{}{
|
||||||
|
{"nodeId": entryNodeID, "protocol": "tls", "strategy": "round"},
|
||||||
|
},
|
||||||
|
"chainNodes": []interface{}{},
|
||||||
|
"outNodeId": []map[string]interface{}{
|
||||||
|
{"nodeId": exitNodeID, "protocol": "tls", "strategy": "round", "port": 43101, "connectIp": "10.41.0.99"},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
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)
|
||||||
|
|
||||||
|
commandMu.Lock()
|
||||||
|
defer commandMu.Unlock()
|
||||||
|
if commandCounts["exit:addservice"] != 2 {
|
||||||
|
t.Fatalf("expected exit AddService twice, got %d (%v)", commandCounts["exit:addservice"], sortedCommandCounts(commandCounts))
|
||||||
|
}
|
||||||
|
if commandCounts["exit:deleteservice"] == 0 {
|
||||||
|
t.Fatalf("expected exit DeleteService retry cleanup to run (%v)", sortedCommandCounts(commandCounts))
|
||||||
|
}
|
||||||
|
if len(addServiceAddrs) < 2 {
|
||||||
|
t.Fatalf("expected recorded AddService addresses, got %v", addServiceAddrs)
|
||||||
|
}
|
||||||
|
for _, addr := range addServiceAddrs {
|
||||||
|
if addr != "10.41.0.99:43101" {
|
||||||
|
t.Fatalf("expected connectIp to stay preferred in AddService addr, got %q", addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func startMockNodeSessionWithCommandFailures(t *testing.T, baseURL string, nodeSecret string, failCommands map[string]string) func() {
|
func startMockNodeSessionWithCommandFailures(t *testing.T, baseURL string, nodeSecret string, failCommands map[string]string) func() {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
@@ -633,3 +762,112 @@ func startMockNodeSessionWithCommandFailures(t *testing.T, baseURL string, nodeS
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func startMockNodeSessionWithCommandRecorder(t *testing.T, baseURL string, nodeSecret string, onCommand func(cmdType string, data json.RawMessage) (bool, string)) func() {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
u, err := url.Parse(baseURL)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("parse provider url: %v", err)
|
||||||
|
}
|
||||||
|
if strings.EqualFold(u.Scheme, "https") {
|
||||||
|
u.Scheme = "wss"
|
||||||
|
} else {
|
||||||
|
u.Scheme = "ws"
|
||||||
|
}
|
||||||
|
u.Path = "/system-info"
|
||||||
|
q := u.Query()
|
||||||
|
q.Set("type", "1")
|
||||||
|
q.Set("secret", nodeSecret)
|
||||||
|
q.Set("version", "v1")
|
||||||
|
q.Set("http", "1")
|
||||||
|
q.Set("tls", "1")
|
||||||
|
q.Set("socks", "1")
|
||||||
|
u.RawQuery = q.Encode()
|
||||||
|
|
||||||
|
conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("dial mock node websocket: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for {
|
||||||
|
_, raw, readErr := conn.ReadMessage()
|
||||||
|
if readErr != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
plain := raw
|
||||||
|
var wrap struct {
|
||||||
|
Encrypted bool `json:"encrypted"`
|
||||||
|
Data string `json:"data"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(raw, &wrap); err == nil && wrap.Encrypted && strings.TrimSpace(wrap.Data) != "" {
|
||||||
|
crypto, cryptoErr := security.NewAESCrypto(nodeSecret)
|
||||||
|
if cryptoErr == nil {
|
||||||
|
if dec, decErr := crypto.Decrypt(wrap.Data); decErr == nil {
|
||||||
|
plain = []byte(dec)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var cmd struct {
|
||||||
|
Type string `json:"type"`
|
||||||
|
RequestID string `json:"requestId"`
|
||||||
|
Data json.RawMessage `json:"data"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(plain, &cmd); err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if strings.TrimSpace(cmd.RequestID) == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
shouldFail := false
|
||||||
|
failMsg := ""
|
||||||
|
if onCommand != nil {
|
||||||
|
shouldFail, failMsg = onCommand(strings.TrimSpace(cmd.Type), cmd.Data)
|
||||||
|
}
|
||||||
|
|
||||||
|
respType := fmt.Sprintf("%sResponse", cmd.Type)
|
||||||
|
respPayload := map[string]interface{}{
|
||||||
|
"type": respType,
|
||||||
|
"success": !shouldFail,
|
||||||
|
"message": "OK",
|
||||||
|
"requestId": cmd.RequestID,
|
||||||
|
}
|
||||||
|
if shouldFail {
|
||||||
|
if strings.TrimSpace(failMsg) == "" {
|
||||||
|
failMsg = "mock command failed"
|
||||||
|
}
|
||||||
|
respPayload["message"] = failMsg
|
||||||
|
}
|
||||||
|
|
||||||
|
respBytes, err := json.Marshal(respPayload)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_ = conn.WriteMessage(websocket.TextMessage, respBytes)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
var stopOnce sync.Once
|
||||||
|
return func() {
|
||||||
|
stopOnce.Do(func() {
|
||||||
|
_ = conn.Close()
|
||||||
|
wg.Wait()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func sortedCommandCounts(counts map[string]int) []string {
|
||||||
|
items := make([]string, 0, len(counts))
|
||||||
|
for key, value := range counts {
|
||||||
|
items = append(items, fmt.Sprintf("%s=%d", key, value))
|
||||||
|
}
|
||||||
|
sort.Strings(items)
|
||||||
|
return items
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,27 @@
|
|||||||
|
# 016 Tunnel Runtime Bind Conflict Retry
|
||||||
|
|
||||||
|
## Checklist
|
||||||
|
|
||||||
|
- [x] Confirm tunnel `connectIp` precedence remains `connectIp > node tcp_listen_addr` for runtime service listen address.
|
||||||
|
- [x] Add tunnel runtime `address already in use` recovery that deletes the stale service and retries `AddService`.
|
||||||
|
- [x] Keep non-bind failures unchanged and avoid altering tunnel chain apply semantics.
|
||||||
|
- [x] Add regression tests for tunnel service address precedence and bind-conflict retry behavior.
|
||||||
|
- [x] Run focused backend handler tests and record the result.
|
||||||
|
- [ ] Add a contract test that simulates node-side `address already in use` during tunnel update and verifies retry success.
|
||||||
|
- [ ] Investigate whether forward update `address already in use` reports are only tunnel-redeploy linkage or also an independent forward path.
|
||||||
|
- [x] Add a contract test that simulates node-side `address already in use` during tunnel update and verifies retry success.
|
||||||
|
- [x] Investigate whether forward update `address already in use` reports are only tunnel-redeploy linkage or also an independent forward path.
|
||||||
|
|
||||||
|
## Test Record
|
||||||
|
|
||||||
|
- Command: `cd go-backend && go test ./internal/http/handler/...`
|
||||||
|
- Result: passed.
|
||||||
|
- Command: `cd go-backend && go test ./tests/contract/... -run 'TestTunnelUpdateRecoversFromAddressInUseContract|TestForwardCreateRollbackWhenServiceDispatchReturnsAddressInUseContract|TestForwardUpdateIgnoresDeletedSpeedLimitContract'`
|
||||||
|
- Result: passed.
|
||||||
|
- Command: `cd go-backend && go test ./tests/contract/... -run 'TestForwardUpdateRecoversFromAddressInUseContract|TestTunnelUpdateRecoversFromAddressInUseContract'`
|
||||||
|
- Result: passed.
|
||||||
|
|
||||||
|
## Investigation Note
|
||||||
|
|
||||||
|
- Forward update still has its own independent `address already in use` recovery path in `syncForwardServicesWithWarnings` / `rebindForwardServiceOnSelfOccupiedPort`; tunnel update linkage is not the only possible source of the symptom.
|
||||||
|
- Tunnel update also triggers downstream forward `UpdateService` for bound forwards, so users can still observe the same error around a tunnel edit even when the failing runtime is on the tunnel side.
|
||||||
Reference in New Issue
Block a user