tcjlj/flvx
1
0
Fork 0
mirror of https://github.com/Sagit-chu/flvx.git synced 2026-10-04 09:06:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(backend): resolve SQLite deadlock in tunnel entry updates (#334)

Browse Source 
- Fix deadlock when updating tunnel entries with offline nodes
- Add test file for tunnel entry SQLite operations
- Update contract tests for entry port conflict and limiter sync
- Add plan documents for SQLite deadlock fix and contract semantics
This commit is contained in:
sagit
2026-03-18 09:00:20 +08:00
committed by GitHub
parent 41ef814643
commit 2df061a19f
7 changed files with 253 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go-backend/internal/http/handler/mutations.go
+21 -7
View File
@@ -832,7 +832,7 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) {
newEntryNodeIDs = append(newEntryNodeIDs, in.NodeID)
}
}
if err := h.validateTunnelEntryPortConflictsForNewEntries(id, oldEntryNodeIDs, newEntryNodeIDs); err != nil {
if err := h.validateTunnelEntryPortConflictsForNewEntries(tx, id, oldEntryNodeIDs, newEntryNodeIDs); err != nil {
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
}
@@ -1020,8 +1020,8 @@ func (h *Handler) cleanupTunnelForwardRuntimesOnRemovedEntryNodes(tunnelID int64
}
}
func (h *Handler) validateTunnelEntryPortConflictsForNewEntries(tunnelID int64, oldEntryNodeIDs, newEntryNodeIDs []int64) error {
if h == nil || h.repo == nil || tunnelID <= 0 {
func (h *Handler) validateTunnelEntryPortConflictsForNewEntries(tx *gorm.DB, tunnelID int64, oldEntryNodeIDs, newEntryNodeIDs []int64) error {
if h == nil || h.repo == nil || tx == nil || tunnelID <= 0 {
return nil
}
@@ -1030,7 +1030,7 @@ func (h *Handler) validateTunnelEntryPortConflictsForNewEntries(tunnelID int64,
return nil
}
forwards, err := h.listForwardsByTunnel(tunnelID)
forwards, err := h.repo.ListForwardsByTunnelTx(tx, tunnelID)
if err != nil || len(forwards) == 0 {
return nil
}
@@ -1040,7 +1040,7 @@ func (h *Handler) validateTunnelEntryPortConflictsForNewEntries(tunnelID int64,
if f == nil {
continue
}
oldPorts, portsErr := h.listForwardPorts(f.ID)
oldPorts, portsErr := h.repo.ListForwardPortsTx(tx, f.ID)
if portsErr != nil {
continue
}
@@ -1050,14 +1050,14 @@ func (h *Handler) validateTunnelEntryPortConflictsForNewEntries(tunnelID int64,
}
for _, nodeID := range addedNodeIDs {
node, nodeErr := h.getNodeRecord(nodeID)
node, nodeErr := h.repo.GetNodeRecordTx(tx, nodeID)
if nodeErr != nil {
continue
}
if err := validateLocalNodePort(node, port); err != nil {
return fmt.Errorf("转发 %s 入口端口冲突: %w", f.Name, err)
}
if err := h.validateForwardPortAvailability(node, port, f.ID); err != nil {
if err := h.validateForwardPortAvailabilityTx(tx, node, port, f.ID); err != nil {
return fmt.Errorf("转发 %s 入口端口冲突: %w", f.Name, err)
}
}
@@ -4153,6 +4153,20 @@ func (h *Handler) validateForwardPortAvailability(node *nodeRecord, port int, cu
return nil
}
func (h *Handler) validateForwardPortAvailabilityTx(tx *gorm.DB, node *nodeRecord, port int, currentForwardID int64) error {
if h == nil || h.repo == nil || tx == nil || node == nil || port <= 0 {
return nil
}
occupied, err := h.repo.HasOtherForwardOnNodePortTx(tx, node.ID, port, currentForwardID)
if err != nil {
return err
}
if occupied {
return fmt.Errorf("节点 %s 端口 %d 已被其他转发占用", node.Name, port)
}
return nil
}
func parsePortRangeMinMax(input string) (int, int) {
input = strings.TrimSpace(input)
if input == "" {
go-backend/internal/http/handler/tunnel_entry_sqlite_test.go
+104
View File
@@ -0,0 +1,104 @@
package handler
import (
"path/filepath"
"testing"
"time"
"go-backend/internal/store/repo"
)
func TestValidateTunnelEntryPortConflictsForNewEntriesDoesNotBlockOnSQLiteTx(t *testing.T) {
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db"))
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
t.Cleanup(func() {
_ = r.Close()
})
h := &Handler{repo: r}
now := time.Now().UnixMilli()
if err := r.DB().Exec(`
INSERT INTO node(name, secret, server_ip, port, created_time, status, tcp_listen_addr, udp_listen_addr, is_remote)
VALUES
('entry-old', 'secret-old', '10.0.0.1', '12000-12010', ?, 1, '[::]', '[::]', 0),
('entry-new', 'secret-new', '10.0.0.2', '12000-12010', ?, 1, '[::]', '[::]', 0)
`, now, now).Error; err != nil {
t.Fatalf("insert nodes: %v", err)
}
var oldEntryID, newEntryID int64
if err := r.DB().Raw(`SELECT id FROM node WHERE name = 'entry-old'`).Scan(&oldEntryID).Error; err != nil {
t.Fatalf("load old entry id: %v", err)
}
if err := r.DB().Raw(`SELECT id FROM node WHERE name = 'entry-new'`).Scan(&newEntryID).Error; err != nil {
t.Fatalf("load new entry id: %v", err)
}
if err := r.DB().Exec(`
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, inx, ip_preference)
VALUES('sqlite-tunnel', 1, 1, 'tls', 1, ?, ?, 1, 1, '')
`, now, now).Error; err != nil {
t.Fatalf("insert tunnel: %v", err)
}
var tunnelID int64
if err := r.DB().Raw(`SELECT id FROM tunnel WHERE name = 'sqlite-tunnel'`).Scan(&tunnelID).Error; err != nil {
t.Fatalf("load tunnel id: %v", err)
}
if err := r.DB().Exec(`
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, inx, protocol)
VALUES(?, '1', ?, 1, 'tls')
`, tunnelID, oldEntryID).Error; err != nil {
t.Fatalf("insert chain_tunnel: %v", err)
}
if err := r.DB().Exec(`
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, created_time, updated_time, status, inx)
VALUES(1, 'tester', 'forward-a', ?, '127.0.0.1:8080', 'fifo', ?, ?, 1, 1)
`, tunnelID, now, now).Error; err != nil {
t.Fatalf("insert forward: %v", err)
}
var forwardID int64
if err := r.DB().Raw(`SELECT id FROM forward WHERE name = 'forward-a'`).Scan(&forwardID).Error; err != nil {
t.Fatalf("load forward id: %v", err)
}
if err := r.DB().Exec(`
INSERT INTO forward_port(forward_id, node_id, port)
VALUES(?, ?, 12001)
`, forwardID, oldEntryID).Error; err != nil {
t.Fatalf("insert forward_port: %v", err)
}
tx := r.BeginTx()
if tx == nil {
t.Fatal("begin tx: nil transaction")
}
if tx.Error != nil {
t.Fatalf("begin tx: %v", tx.Error)
}
errCh := make(chan error, 1)
doneCh := make(chan struct{})
go func() {
defer close(doneCh)
errCh <- h.validateTunnelEntryPortConflictsForNewEntries(tx, tunnelID, []int64{oldEntryID}, []int64{oldEntryID, newEntryID})
}()
select {
case err := <-errCh:
if err != nil {
_ = tx.Rollback().Error
t.Fatalf("unexpected validation error: %v", err)
}
case <-time.After(500 * time.Millisecond):
_ = tx.Rollback().Error
<-doneCh
t.Fatal("validation blocked while transaction was open on sqlite")
}
if err := tx.Rollback().Error; err != nil {
t.Fatalf("rollback tx: %v", err)
}
}