mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-11-15 17:14:23 +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)
|
_, services, applyErr := h.applyTunnelRuntimeUpsert(state)
|
||||||
applyErr = errors.Join(applyErr, h.acknowledgeTunnelServiceReplacements(tunnelID, services))
|
applyErr = errors.Join(applyErr, h.acknowledgeTunnelServiceReplacements(tunnelID, services))
|
||||||
|
applyErr = errors.Join(applyErr, h.retryTunnelRuntimeCleanupLocked(tunnelID))
|
||||||
if applyErr != nil {
|
if applyErr != nil {
|
||||||
return applyErr
|
return applyErr
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ func (h *Handler) queueObsoleteTunnelRuntimeTx(tx *gorm.DB, tunnelID int64, oldR
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
for _, current := range newRows {
|
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})
|
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{}
|
liveServices := map[int64]bool{}
|
||||||
for _, item := range items {
|
for _, item := range items {
|
||||||
desired := false
|
desired := false
|
||||||
desiredPort := 0
|
|
||||||
for _, row := range rows {
|
for _, row := range rows {
|
||||||
if row.NodeID == item.NodeID && ((item.Kind == "service" && tunnelRuntimeNeedsService(row)) || (item.Kind == "chain" && tunnelRuntimeNeedsChain(row))) {
|
if row.NodeID == item.NodeID && ((item.Kind == "service" && tunnelRuntimeNeedsService(row)) || (item.Kind == "chain" && tunnelRuntimeNeedsChain(row))) {
|
||||||
desired = true
|
desired = true
|
||||||
desiredPort = row.Port
|
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if desired && item.Kind == "service" && desiredPort != item.Port {
|
if desired && item.Kind == "service" {
|
||||||
// Same service name, changed port: keep the old reservation until an
|
// Desired state does not prove that the old receiver was replaced.
|
||||||
// acknowledged upsert proves that the old listener has been replaced.
|
// Keep this barrier (even at the same port) and protect its old
|
||||||
|
// chain until acknowledgeTunnelServiceReplacements observes ACK.
|
||||||
|
liveServices[item.NodeID] = true
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if !desired {
|
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.
|
- `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.
|
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