From 87479c2ac1b5a52a28f326b213d5a7f6278891d9 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Sat, 7 Mar 2026 16:21:44 +0800 Subject: [PATCH] fix: add retry mechanism for tunnel service bind conflicts When tunnel services encounter 'address already in use' errors during creation/update, automatically cleanup and retry once instead of failing immediately. This handles race conditions during rapid tunnel reconfiguration. Entire-Checkpoint: 39e6fb9de836 --- .../http/handler/control_plane_test.go | 69 +++++ .../internal/http/handler/dual_stack_test.go | 13 + go-backend/internal/http/handler/mutations.go | 44 +++- .../tests/contract/forward_contract_test.go | 160 ++++++++++++ .../limiter_sync_failure_contract_test.go | 238 ++++++++++++++++++ .../016-tunnel-runtime-bind-conflict-retry.md | 27 ++ 6 files changed, 549 insertions(+), 2 deletions(-) create mode 100644 plans/016-tunnel-runtime-bind-conflict-retry.md diff --git a/go-backend/internal/http/handler/control_plane_test.go b/go-backend/internal/http/handler/control_plane_test.go index 1663af1..55a6da5 100644 --- a/go-backend/internal/http/handler/control_plane_test.go +++ b/go-backend/internal/http/handler/control_plane_test.go @@ -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) { forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"} diff --git a/go-backend/internal/http/handler/dual_stack_test.go b/go-backend/internal/http/handler/dual_stack_test.go index 476f1ce..d25e825 100644 --- a/go-backend/internal/http/handler/dual_stack_test.go +++ b/go-backend/internal/http/handler/dual_stack_test.go @@ -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) { node := &nodeRecord{TCPListenAddr: "[::]"} chain := tunnelRuntimeNode{Protocol: "tls", Port: 21001} diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index f5cae8f..5fcddf1 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -25,6 +25,8 @@ import ( "gorm.io/gorm" ) +const tunnelServiceBindRetryDelay = 150 * time.Millisecond + func (h *Handler) userCreate(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { response.WriteJSON(w, response.ErrDefault("请求失败")) @@ -2584,7 +2586,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 createdChains = append(createdChains, 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) } createdServices = append(createdServices, chainNode.NodeID) @@ -2596,7 +2598,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 continue } 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) } createdServices = append(createdServices, outNode.NodeID) @@ -2605,6 +2607,44 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 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) { if h == nil || tunnelID <= 0 { return diff --git a/go-backend/tests/contract/forward_contract_test.go b/go-backend/tests/contract/forward_contract_test.go index 9c19c4f..c2dd0de 100644 --- a/go-backend/tests/contract/forward_contract_test.go +++ b/go-backend/tests/contract/forward_contract_test.go @@ -7,6 +7,8 @@ import ( "net/http" "net/http/httptest" "strconv" + "strings" + "sync" "testing" "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 { return strconv.FormatInt(v, 10) } diff --git a/go-backend/tests/contract/limiter_sync_failure_contract_test.go b/go-backend/tests/contract/limiter_sync_failure_contract_test.go index 88bbda5..7967b5d 100644 --- a/go-backend/tests/contract/limiter_sync_failure_contract_test.go +++ b/go-backend/tests/contract/limiter_sync_failure_contract_test.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "sort" "strings" "sync" "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() { 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 +} diff --git a/plans/016-tunnel-runtime-bind-conflict-retry.md b/plans/016-tunnel-runtime-bind-conflict-retry.md new file mode 100644 index 0000000..b737a00 --- /dev/null +++ b/plans/016-tunnel-runtime-bind-conflict-retry.md @@ -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.