tcjlj/flvx
1
0
Fork 0
mirror of https://github.com/Sagit-chu/flvx.git synced 2026-09-30 16:26:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: rollback forward mutation on tunnel switch failure

Browse Source
This commit is contained in:
sagit
2026-02-08 02:30:36 +00:00
parent d720b9e00c
commit 7f9c05172b
3 changed files with 355 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go-backend/internal/http/handler/control_plane.go
+34 -1
View File
@@ -283,6 +283,32 @@ func (h *Handler) listUserTunnelIDs(userID, tunnelID int64) ([]int64, error) {
return out, nil
}
func (h *Handler) listUserTunnelIDsByUser(userID int64) ([]int64, error) {
rows, err := h.repo.DB().Query(`
SELECT id
FROM user_tunnel
WHERE user_id = ?
ORDER BY id ASC
`, userID)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]int64, 0)
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
return nil, err
}
out = append(out, id)
}
if err := rows.Err(); err != nil {
return nil, err
}
return out, nil
}
func (h *Handler) syncForwardServices(forward *forwardRecord, method string, allowFallbackAdd bool) error {
if h == nil || forward == nil {
return errors.New("invalid forward sync context")
@@ -342,7 +368,14 @@ func (h *Handler) controlForwardServices(forward *forwardRecord, commandType str
if err != nil {
return err
}
bases := buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, userTunnelIDs)
allUserTunnelIDs, err := h.listUserTunnelIDsByUser(forward.UserID)
if err != nil {
return err
}
candidateTunnelIDs := make([]int64, 0, len(userTunnelIDs)+len(allUserTunnelIDs))
candidateTunnelIDs = append(candidateTunnelIDs, userTunnelIDs...)
candidateTunnelIDs = append(candidateTunnelIDs, allUserTunnelIDs...)
bases := buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, candidateTunnelIDs)
seen := map[int64]struct{}{}
for _, fp := range ports {
if _, ok := seen[fp.NodeID]; ok {
go-backend/internal/http/handler/mutations.go
+71 -5
View File
@@ -948,6 +948,11 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
oldPorts, err := h.listForwardPorts(id)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
tunnelID := asInt64(req["tunnelId"], forward.TunnelID)
if tunnelID <= 0 {
@@ -1000,13 +1005,19 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
_ = h.replaceForwardPorts(id, tunnelID, port)
if err := h.replaceForwardPorts(id, tunnelID, port); err != nil {
h.rollbackForwardMutation(forward, oldPorts)
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
updatedForward, err := h.getForwardRecord(id)
if err != nil {
h.rollbackForwardMutation(forward, oldPorts)
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
if err := h.syncForwardServices(updatedForward, "UpdateService", true); err != nil {
h.rollbackForwardMutation(forward, oldPorts)
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
}
@@ -1291,6 +1302,11 @@ func (h *Handler) forwardBatchChangeTunnel(w http.ResponseWriter, r *http.Reques
fail++
continue
}
oldPorts, listPortsErr := h.listForwardPorts(id)
if listPortsErr != nil {
fail++
continue
}
var port sql.NullInt64
_ = h.repo.DB().QueryRow(`SELECT MIN(port) FROM forward_port WHERE forward_id = ?`, id).Scan(&port)
_, err := h.repo.DB().Exec(`UPDATE forward SET tunnel_id = ?, updated_time = ? WHERE id = ?`, req.TargetTunnelID, time.Now().UnixMilli(), id)
@@ -1305,13 +1321,19 @@ func (h *Handler) forwardBatchChangeTunnel(w http.ResponseWriter, r *http.Reques
if p <= 0 {
p = h.pickTunnelPort(req.TargetTunnelID)
}
_ = h.replaceForwardPorts(id, req.TargetTunnelID, p)
if err := h.replaceForwardPorts(id, req.TargetTunnelID, p); err != nil {
h.rollbackForwardMutation(forward, oldPorts)
fail++
continue
}
updatedForward, fetchErr := h.getForwardRecord(id)
if fetchErr != nil {
h.rollbackForwardMutation(forward, oldPorts)
fail++
continue
}
if err := h.syncForwardServices(updatedForward, "UpdateService", true); err != nil {
h.rollbackForwardMutation(forward, oldPorts)
fail++
continue
}
@@ -2407,14 +2429,58 @@ func (h *Handler) replaceForwardPorts(forwardID, tunnelID int64, port int) error
return err
}
defer func() { _ = tx.Rollback() }()
_, _ = tx.Exec(`DELETE FROM forward_port WHERE forward_id = ?`, forwardID)
entryNodes, _ := h.tunnelEntryNodeIDs(tunnelID)
if _, err := tx.Exec(`DELETE FROM forward_port WHERE forward_id = ?`, forwardID); err != nil {
return err
}
entryNodes, err := h.tunnelEntryNodeIDs(tunnelID)
if err != nil {
return err
}
for _, nodeID := range entryNodes {
_, _ = tx.Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, nodeID, port)
if _, err := tx.Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, nodeID, port); err != nil {
return err
}
}
return tx.Commit()
}
func (h *Handler) replaceForwardPortsWithRecords(forwardID int64, ports []forwardPortRecord) error {
tx, err := h.repo.DB().Begin()
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
if _, err := tx.Exec(`DELETE FROM forward_port WHERE forward_id = ?`, forwardID); err != nil {
return err
}
for _, fp := range ports {
if _, err := tx.Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, fp.NodeID, fp.Port); err != nil {
return err
}
}
return tx.Commit()
}
func (h *Handler) rollbackForwardMutation(oldForward *forwardRecord, oldPorts []forwardPortRecord) {
if h == nil || oldForward == nil || h.repo == nil || h.repo.DB() == nil {
return
}
_, _ = h.repo.DB().Exec(`
UPDATE forward
SET user_id = ?, user_name = ?, name = ?, tunnel_id = ?, remote_addr = ?, strategy = ?, status = ?, updated_time = ?
WHERE id = ?
`, oldForward.UserID, oldForward.UserName, oldForward.Name, oldForward.TunnelID, oldForward.RemoteAddr, oldForward.Strategy, oldForward.Status, time.Now().UnixMilli(), oldForward.ID)
if err := h.replaceForwardPortsWithRecords(oldForward.ID, oldPorts); err != nil {
return
}
_ = h.syncForwardServices(oldForward, "UpdateService", true)
}
func (h *Handler) upsertUserTunnel(req map[string]interface{}) error {
userID := asInt64(req["userId"], 0)
tunnelID := asInt64(req["tunnelId"], 0)
go-backend/tests/contract/forward_contract_test.go
+250
View File
@@ -197,6 +197,256 @@ func TestForwardOwnershipAndScopeContracts(t *testing.T) {
})
}
func TestForwardSwitchTunnelRollbackOnSyncFailure(t *testing.T) {
secret := "contract-jwt-secret"
router, repo := setupContractRouter(t, secret)
now := time.Now().UnixMilli()
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
if err != nil {
t.Fatalf("generate admin token: %v", err)
}
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(2, 'switch_user', '3c85cdebade1c51cf64ca9f3c09d182d', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
`, now, now); err != nil {
t.Fatalf("insert user: %v", err)
}
insertTunnel := func(name string, inx int) int64 {
res, err := repo.DB().Exec(`
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, name, 1.0, 1, "tls", 99999, now, now, 1, nil, inx)
if err != nil {
t.Fatalf("insert tunnel %s: %v", name, err)
}
id, err := res.LastInsertId()
if err != nil {
t.Fatalf("get tunnel id %s: %v", name, err)
}
return id
}
insertNode := func(name, ip, portRange string, inx int) int64 {
res, 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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, name, name+"-secret", ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", inx)
if err != nil {
t.Fatalf("insert node %s: %v", name, err)
}
id, err := res.LastInsertId()
if err != nil {
t.Fatalf("get node id %s: %v", name, err)
}
return id
}
tunnelA := insertTunnel("switch-tunnel-a", 0)
tunnelB := insertTunnel("switch-tunnel-b", 1)
nodeA := insertNode("switch-node-a", "10.10.0.1", "21000-21010", 0)
nodeB := insertNode("switch-node-b", "10.10.0.2", "22000-22010", 1)
if _, err := repo.DB().Exec(`
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
VALUES(?, 1, ?, 21001, 'round', 1, 'tls')
`, tunnelA, nodeA); err != nil {
t.Fatalf("insert chain_tunnel tunnelA: %v", err)
}
if _, err := repo.DB().Exec(`
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
VALUES(?, 1, ?, 22001, 'round', 1, 'tls')
`, tunnelB, nodeB); err != nil {
t.Fatalf("insert chain_tunnel tunnelB: %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(10, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
`, tunnelA); err != nil {
t.Fatalf("insert user_tunnel A: %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(11, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
`, tunnelB); err != nil {
t.Fatalf("insert user_tunnel B: %v", err)
}
forwardRes, err := repo.DB().Exec(`
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
VALUES(2, 'switch_user', 'switch-forward', ?, '8.8.8.8:53', 'fifo', 0, 0, ?, ?, 1, 0)
`, tunnelA, now, now)
if err != nil {
t.Fatalf("insert forward: %v", err)
}
forwardID, err := forwardRes.LastInsertId()
if err != nil {
t.Fatalf("get forward id: %v", err)
}
if _, err := repo.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, nodeA, 21001); err != nil {
t.Fatalf("insert forward_port: %v", err)
}
payload := `{"id":` + jsonNumber(forwardID) + `,"tunnelId":` + jsonNumber(tunnelB) + `}`
req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/update", bytes.NewBufferString(payload))
req.Header.Set("Authorization", adminToken)
req.Header.Set("Content-Type", "application/json")
res := httptest.NewRecorder()
router.ServeHTTP(res, req)
var out response.R
if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
t.Fatalf("decode response: %v", err)
}
if out.Code == 0 {
t.Fatalf("expected update failure when node is offline")
}
var tunnelAfter int64
if err := repo.DB().QueryRow(`SELECT tunnel_id FROM forward WHERE id = ?`, forwardID).Scan(&tunnelAfter); err != nil {
t.Fatalf("query forward tunnel_id: %v", err)
}
if tunnelAfter != tunnelA {
t.Fatalf("expected tunnel rollback to %d, got %d", tunnelA, tunnelAfter)
}
var nodeAfter int64
var portAfter int
if err := repo.DB().QueryRow(`SELECT node_id, port FROM forward_port WHERE forward_id = ? LIMIT 1`, forwardID).Scan(&nodeAfter, &portAfter); err != nil {
t.Fatalf("query forward_port: %v", err)
}
if nodeAfter != nodeA || portAfter != 21001 {
t.Fatalf("expected forward_port rollback to node=%d port=21001, got node=%d port=%d", nodeA, nodeAfter, portAfter)
}
}
func TestForwardBatchChangeTunnelRollbackOnSyncFailure(t *testing.T) {
secret := "contract-jwt-secret"
router, repo := setupContractRouter(t, secret)
now := time.Now().UnixMilli()
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
if err != nil {
t.Fatalf("generate admin token: %v", err)
}
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(2, 'batch_switch_user', '3c85cdebade1c51cf64ca9f3c09d182d', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
`, now, now); err != nil {
t.Fatalf("insert user: %v", err)
}
tunnelResA, err := repo.DB().Exec(`
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
VALUES('batch-switch-tunnel-a', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0)
`, now, now)
if err != nil {
t.Fatalf("insert tunnel A: %v", err)
}
tunnelA, _ := tunnelResA.LastInsertId()
tunnelResB, err := repo.DB().Exec(`
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
VALUES('batch-switch-tunnel-b', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 1)
`, now, now)
if err != nil {
t.Fatalf("insert tunnel B: %v", err)
}
tunnelB, _ := tunnelResB.LastInsertId()
nodeResA, 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('batch-switch-node-a', 'batch-switch-node-a-secret', '10.11.0.1', '10.11.0.1', '', '23000-23010', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 0)
`, now, now)
if err != nil {
t.Fatalf("insert node A: %v", err)
}
nodeA, _ := nodeResA.LastInsertId()
nodeResB, 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('batch-switch-node-b', 'batch-switch-node-b-secret', '10.11.0.2', '10.11.0.2', '', '24000-24010', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 1)
`, now, now)
if err != nil {
t.Fatalf("insert node B: %v", err)
}
nodeB, _ := nodeResB.LastInsertId()
if _, err := repo.DB().Exec(`INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) VALUES(?, 1, ?, 23001, 'round', 1, 'tls')`, tunnelA, nodeA); err != nil {
t.Fatalf("insert chain_tunnel A: %v", err)
}
if _, err := repo.DB().Exec(`INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) VALUES(?, 1, ?, 24001, 'round', 1, 'tls')`, tunnelB, nodeB); err != nil {
t.Fatalf("insert chain_tunnel B: %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(20, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)`, tunnelA); err != nil {
t.Fatalf("insert user_tunnel A: %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(21, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)`, tunnelB); err != nil {
t.Fatalf("insert user_tunnel B: %v", err)
}
forwardRes, err := repo.DB().Exec(`
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
VALUES(2, 'batch_switch_user', 'batch-switch-forward', ?, '1.1.1.1:443', 'fifo', 0, 0, ?, ?, 1, 0)
`, tunnelA, now, now)
if err != nil {
t.Fatalf("insert forward: %v", err)
}
forwardID, _ := forwardRes.LastInsertId()
if _, err := repo.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, nodeA, 23001); err != nil {
t.Fatalf("insert forward_port: %v", err)
}
payload := `{"forwardIds":[` + jsonNumber(forwardID) + `],"targetTunnelId":` + jsonNumber(tunnelB) + `}`
req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/batch-change-tunnel", bytes.NewBufferString(payload))
req.Header.Set("Authorization", adminToken)
req.Header.Set("Content-Type", "application/json")
res := httptest.NewRecorder()
router.ServeHTTP(res, req)
var out response.R
if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
t.Fatalf("decode response: %v", err)
}
if out.Code != 0 {
t.Fatalf("expected API success envelope, got code=%d msg=%q", out.Code, out.Msg)
}
result, ok := out.Data.(map[string]interface{})
if !ok {
t.Fatalf("expected map result, got %T", out.Data)
}
if int(result["failCount"].(float64)) != 1 {
t.Fatalf("expected failCount=1, got %v", result["failCount"])
}
var tunnelAfter int64
if err := repo.DB().QueryRow(`SELECT tunnel_id FROM forward WHERE id = ?`, forwardID).Scan(&tunnelAfter); err != nil {
t.Fatalf("query forward tunnel_id: %v", err)
}
if tunnelAfter != tunnelA {
t.Fatalf("expected tunnel rollback to %d, got %d", tunnelA, tunnelAfter)
}
var nodeAfter int64
var portAfter int
if err := repo.DB().QueryRow(`SELECT node_id, port FROM forward_port WHERE forward_id = ? LIMIT 1`, forwardID).Scan(&nodeAfter, &portAfter); err != nil {
t.Fatalf("query forward_port: %v", err)
}
if nodeAfter != nodeA || portAfter != 23001 {
t.Fatalf("expected forward_port rollback to node=%d port=23001, got node=%d port=%d", nodeA, nodeAfter, portAfter)
}
}
func jsonNumber(v int64) string {
return strconv.FormatInt(v, 10)
}