From 89a1a7bff3bfc18f4b27fb9b1d9398353b4d32a4 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Sun, 8 Nov 2026 23:36:43 +0800 Subject: [PATCH] fix(tunnel): retain chains until receiver replacement is acknowledged --- go-backend/internal/http/handler/mutations.go | 1 + .../http/handler/tunnel_runtime_cleanup.go | 12 +- .../contract/tunnel_role_transition_test.go | 117 ++++++++++++++++++ .../070-tunnel-lifecycle-validation.md | 4 + 4 files changed, 128 insertions(+), 6 deletions(-) create mode 100644 go-backend/tests/contract/tunnel_role_transition_test.go diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 9a8cfc2..75947a2 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -1973,6 +1973,7 @@ func (h *Handler) redeployTunnelAndForwardsLocked(tunnelID int64) error { } _, services, applyErr := h.applyTunnelRuntimeUpsert(state) applyErr = errors.Join(applyErr, h.acknowledgeTunnelServiceReplacements(tunnelID, services)) + applyErr = errors.Join(applyErr, h.retryTunnelRuntimeCleanupLocked(tunnelID)) if applyErr != nil { return applyErr } diff --git a/go-backend/internal/http/handler/tunnel_runtime_cleanup.go b/go-backend/internal/http/handler/tunnel_runtime_cleanup.go index d3ff7ae..4e6af6f 100644 --- a/go-backend/internal/http/handler/tunnel_runtime_cleanup.go +++ b/go-backend/internal/http/handler/tunnel_runtime_cleanup.go @@ -42,7 +42,7 @@ func (h *Handler) queueObsoleteTunnelRuntimeTx(tx *gorm.DB, tunnelID int64, oldR continue } for _, current := range newRows { - if current.NodeID == old.NodeID && tunnelRuntimeNeedsService(current) && current.Port != old.Port { + if current.NodeID == old.NodeID && tunnelRuntimeNeedsService(current) && (current.Port != old.Port || current.ChainType != old.ChainType) { items = append(items, model.TunnelRuntimeCleanup{TunnelID: tunnelID, NodeID: old.NodeID, Kind: "service", Port: old.Port}) } } @@ -78,17 +78,17 @@ func (h *Handler) retryTunnelRuntimeCleanupLocked(tunnelID int64) error { liveServices := map[int64]bool{} for _, item := range items { desired := false - desiredPort := 0 for _, row := range rows { if row.NodeID == item.NodeID && ((item.Kind == "service" && tunnelRuntimeNeedsService(row)) || (item.Kind == "chain" && tunnelRuntimeNeedsChain(row))) { desired = true - desiredPort = row.Port break } } - if desired && item.Kind == "service" && desiredPort != item.Port { - // Same service name, changed port: keep the old reservation until an - // acknowledged upsert proves that the old listener has been replaced. + if desired && item.Kind == "service" { + // Desired state does not prove that the old receiver was replaced. + // Keep this barrier (even at the same port) and protect its old + // chain until acknowledgeTunnelServiceReplacements observes ACK. + liveServices[item.NodeID] = true continue } if !desired { diff --git a/go-backend/tests/contract/tunnel_role_transition_test.go b/go-backend/tests/contract/tunnel_role_transition_test.go new file mode 100644 index 0000000..fd2f3a5 --- /dev/null +++ b/go-backend/tests/contract/tunnel_role_transition_test.go @@ -0,0 +1,117 @@ +package contract_test + +import ( + "bytes" + "encoding/json" + "fmt" + "go-backend/internal/auth" + "net/http" + "net/http/httptest" + "sync" + "testing" +) + +func TestMiddleToExitKeepsChainUntilReceiverReplacementAcknowledged(t *testing.T) { + for _, port := range []int{42000, 42001} { + for _, failure := range []string{"等待节点响应超时", "replacement rejected"} { + t.Run(fmt.Sprintf("port=%d/failure=%s", port, failure), func(t *testing.T) { + router, r := setupContractRouter(t, "role-transition") + server := httptest.NewServer(router) + defer server.Close() + token, err := auth.GenerateToken(1, "admin", 0, "role-transition") + if err != nil { + t.Fatal(err) + } + entry := insertContractNode(t, r, "role-entry", "192.0.2.10", "41000-41010", "role-entry-secret", 1) + middle := insertContractNode(t, r, "role-middle", "192.0.2.20", "42000-42010", "role-middle-secret", 1) + exit := insertContractNode(t, r, "role-exit", "192.0.2.30", "43000-43010", "role-exit-secret", 1) + stopEntry := startMockNodeSession(t, server.URL, "role-entry-secret") + defer stopEntry() + var mu sync.Mutex + failReplacement := false + replacementAccepted := false + unsafeDeletion := false + chainDeletes := 0 + stopMiddle := startMockNodeSessionWithCommandRecorder(t, server.URL, "role-middle-secret", func(kind string, data json.RawMessage) (bool, string) { + mu.Lock() + defer mu.Unlock() + if kind == "UpdateService" { + if failReplacement { + return true, failure + } + var services []struct { + Handler struct { + Chain string `json:"chain"` + } `json:"handler"` + } + if err := json.Unmarshal(data, &services); err == nil && len(services) > 0 && services[0].Handler.Chain == "" { + replacementAccepted = true + } + } + if kind == "DeleteChains" { + chainDeletes++ + if !replacementAccepted { + unsafeDeletion = true + } + } + return false, "" + }) + defer stopMiddle() + stopExit := startMockNodeSession(t, server.URL, "role-exit-secret") + defer stopExit() + post := func(path string, payload interface{}) *httptest.ResponseRecorder { + raw, _ := json.Marshal(payload) + req := httptest.NewRequest(http.MethodPost, path, bytes.NewReader(raw)) + req.Header.Set("Authorization", token) + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + return res + } + payload := map[string]interface{}{"name": "role-transition", "type": 2, "inNodeId": []map[string]interface{}{{"nodeId": entry, "protocol": "tls"}}, "chainNodes": [][]map[string]interface{}{{{"nodeId": middle, "protocol": "tls", "port": 42000}}}, "outNodeId": []map[string]interface{}{{"nodeId": exit, "protocol": "tls", "port": 43000}}} + assertCode(t, post("/api/v1/tunnel/create", payload), 0) + listed := post("/api/v1/tunnel/list", map[string]interface{}{}) + var list struct { + Data []struct { + ID int64 `json:"id"` + } `json:"data"` + } + if err := json.Unmarshal(listed.Body.Bytes(), &list); err != nil || len(list.Data) != 1 { + t.Fatal("missing tunnel") + } + id := list.Data[0].ID + mu.Lock() + failReplacement = true + mu.Unlock() + payload["id"] = id + payload["chainNodes"] = [][]map[string]interface{}{} + payload["outNodeId"] = []map[string]interface{}{{"nodeId": middle, "protocol": "tls", "port": port}} + post("/api/v1/tunnel/update", payload) + post("/api/v1/tunnel/batch-redeploy", map[string]interface{}{"ids": []int64{id}}) + mu.Lock() + if chainDeletes != 0 || unsafeDeletion { + t.Errorf("old middle chain removed before receiver replacement ACK: deletes=%d unsafe=%v", chainDeletes, unsafeDeletion) + } + failReplacement = false + mu.Unlock() + recovered := post("/api/v1/tunnel/batch-redeploy", map[string]interface{}{"ids": []int64{id}}) + var result struct { + Data struct { + SuccessCount int `json:"successCount"` + FailCount int `json:"failCount"` + } `json:"data"` + } + if err := json.Unmarshal(recovered.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + if result.Data.SuccessCount != 1 || result.Data.FailCount != 0 { + t.Fatalf("replacement retry failed: %s", recovered.Body.String()) + } + mu.Lock() + defer mu.Unlock() + if !replacementAccepted || chainDeletes != 1 || unsafeDeletion { + t.Fatalf("obsolete chain must be removed exactly after successful exit replacement: accepted=%v deletes=%d unsafe=%v", replacementAccepted, chainDeletes, unsafeDeletion) + } + }) + } + } +} diff --git a/plans/evidence/070-tunnel-lifecycle-validation.md b/plans/evidence/070-tunnel-lifecycle-validation.md index d37178d..a183b2e 100644 --- a/plans/evidence/070-tunnel-lifecycle-validation.md +++ b/plans/evidence/070-tunnel-lifecycle-validation.md @@ -62,3 +62,7 @@ Public-seam regressions: - `TestDeletedTunnelCleanupSurvivesHandlerReplacement` deletes desired state despite a rejected listener deletion, verifies old-port reservation, then replaces the Handler and reconnects node sessions. Persisted ownership permits cleanup/retry and releases the port only after ACK. Subsequent retry does not remove the replacement tunnel. The full repository, handler and contract suites passed (2.965s, 12.463s, 16.974s) before the final reconnect-specific test and forward-port reservation check. Focused race validation results are reported with the integration checkpoint. + +### Receiver role-transition acknowledgment barrier + +Final review on rebased baseline `7f01774` found that middle→exit on the same node could remove the old chain after a rejected/timed-out receiver replacement. `TestMiddleToExitKeepsChainUntilReceiverReplacementAcknowledged` reproduced this in all four combinations of same/changed receiver port and timeout/hard rejection (`deletes=1 unsafe=true`). The fix queues a service-replacement barrier on receiver role changes, including same-port changes. A still-desired service record is retained and protects chain cleanup until successful service upsert acknowledgment; redeploy drains obsolete chains after that acknowledgment. Focused green result: PASS, 0.764s. Red evidence: `/tmp/flvx-role-transition-red.log`; race validation is reported in the final integration checkpoint.