From 7f9c05172b1ac967f29235b00b52d2e13ae4c675 Mon Sep 17 00:00:00 2001 From: sagit Date: Sun, 8 Feb 2026 02:30:36 +0000 Subject: [PATCH] fix: rollback forward mutation on tunnel switch failure --- .../internal/http/handler/control_plane.go | 35 ++- go-backend/internal/http/handler/mutations.go | 76 +++++- .../tests/contract/forward_contract_test.go | 250 ++++++++++++++++++ 3 files changed, 355 insertions(+), 6 deletions(-) diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 95046b7..a2f0887 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -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 { diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 3569756..d48bb13 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -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) diff --git a/go-backend/tests/contract/forward_contract_test.go b/go-backend/tests/contract/forward_contract_test.go index 7868eb9..88f2f7d 100644 --- a/go-backend/tests/contract/forward_contract_test.go +++ b/go-backend/tests/contract/forward_contract_test.go @@ -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) }