mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-11-13 00:14:22 +08:00
fix(tunnel): retain chains until receiver replacement is acknowledged
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user