From 2df061a19f69aff74aaf7c0b66cb5a3a19b95bbd Mon Sep 17 00:00:00 2001 From: sagit <36596628+Sagit-chu@users.noreply.github.com> Date: Wed, 18 Mar 2026 09:00:20 +0800 Subject: [PATCH] fix(backend): resolve SQLite deadlock in tunnel entry updates (#334) - 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 --- go-backend/internal/http/handler/mutations.go | 28 +++-- .../http/handler/tunnel_entry_sqlite_test.go | 104 ++++++++++++++++++ .../internal/store/repo/repository_control.go | 27 ++++- ...ue313_entry_port_conflict_contract_test.go | 7 +- .../limiter_sync_failure_contract_test.go | 6 +- ...042-sqlite-tunnel-entry-update-deadlock.md | 49 +++++++++ .../043-entry-contract-semantics-alignment.md | 49 +++++++++ 7 files changed, 253 insertions(+), 17 deletions(-) create mode 100644 go-backend/internal/http/handler/tunnel_entry_sqlite_test.go create mode 100644 plans/042-sqlite-tunnel-entry-update-deadlock.md create mode 100644 plans/043-entry-contract-semantics-alignment.md diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 4bf32c8..ecd6e83 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -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 == "" { diff --git a/go-backend/internal/http/handler/tunnel_entry_sqlite_test.go b/go-backend/internal/http/handler/tunnel_entry_sqlite_test.go new file mode 100644 index 0000000..67f4793 --- /dev/null +++ b/go-backend/internal/http/handler/tunnel_entry_sqlite_test.go @@ -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) + } +} diff --git a/go-backend/internal/store/repo/repository_control.go b/go-backend/internal/store/repo/repository_control.go index 7bc0922..894a7f3 100644 --- a/go-backend/internal/store/repo/repository_control.go +++ b/go-backend/internal/store/repo/repository_control.go @@ -30,8 +30,15 @@ func (r *Repository) ListForwardsByTunnel(tunnelID int64) ([]model.ForwardRecord if r == nil || r.db == nil { return nil, errors.New("repository not initialized") } + return r.ListForwardsByTunnelTx(r.db, tunnelID) +} + +func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]model.ForwardRecord, error) { + if tx == nil { + return nil, errors.New("database unavailable") + } var forwards []model.Forward - err := r.db.Where("tunnel_id = ?", tunnelID).Order("id ASC").Find(&forwards).Error + err := tx.Where("tunnel_id = ?", tunnelID).Order("id ASC").Find(&forwards).Error if err != nil { return nil, err } @@ -95,8 +102,15 @@ func (r *Repository) ListForwardPorts(forwardID int64) ([]model.ForwardPortRecor if r == nil || r.db == nil { return nil, errors.New("repository not initialized") } + return r.ListForwardPortsTx(r.db, forwardID) +} + +func (r *Repository) ListForwardPortsTx(tx *gorm.DB, forwardID int64) ([]model.ForwardPortRecord, error) { + if tx == nil { + return nil, errors.New("database unavailable") + } var ports []model.ForwardPort - err := r.db.Where("forward_id = ?", forwardID).Order("id ASC").Find(&ports).Error + err := tx.Where("forward_id = ?", forwardID).Order("id ASC").Find(&ports).Error if err != nil { return nil, err } @@ -115,12 +129,19 @@ func (r *Repository) HasOtherForwardOnNodePort(nodeID int64, port int, currentFo if r == nil || r.db == nil { return false, errors.New("repository not initialized") } + return r.HasOtherForwardOnNodePortTx(r.db, nodeID, port, currentForwardID) +} + +func (r *Repository) HasOtherForwardOnNodePortTx(tx *gorm.DB, nodeID int64, port int, currentForwardID int64) (bool, error) { + if tx == nil { + return false, errors.New("database unavailable") + } if nodeID <= 0 || port <= 0 { return false, nil } var count int64 - err := r.db.Model(&model.ForwardPort{}). + err := tx.Model(&model.ForwardPort{}). Where("node_id = ? AND port = ? AND forward_id <> ?", nodeID, port, currentForwardID). Count(&count).Error if err != nil { diff --git a/go-backend/tests/contract/issue313_entry_port_conflict_contract_test.go b/go-backend/tests/contract/issue313_entry_port_conflict_contract_test.go index 0d57aaf..5ed8d44 100644 --- a/go-backend/tests/contract/issue313_entry_port_conflict_contract_test.go +++ b/go-backend/tests/contract/issue313_entry_port_conflict_contract_test.go @@ -32,7 +32,6 @@ func TestIssue313_EntryPortCrossTunnelConflictContract(t *testing.T) { return mustLastInsertID(t, repo, name) } - entryA := insertNode("issue313-entry-a", "10.100.0.1", "2000-2010") entryB1 := insertNode("issue313-entry-b1", "10.100.0.2", "2000-2010") entryB2 := insertNode("issue313-entry-b2", "10.100.0.3", "2000-2010") chainA := insertNode("issue313-chain-a", "10.100.0.4", "3000-3010") @@ -51,7 +50,7 @@ func TestIssue313_EntryPortCrossTunnelConflictContract(t *testing.T) { if err := repo.DB().Exec(` INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) VALUES(?, 1, ?, 2000, 'round', 1, 'tls') - `, tunnelAID, entryA).Error; err != nil { + `, tunnelAID, entryB2).Error; err != nil { t.Fatalf("insert chain_tunnel entry a: %v", err) } if err := repo.DB().Exec(` @@ -109,7 +108,7 @@ func TestIssue313_EntryPortCrossTunnelConflictContract(t *testing.T) { } forwardAID := mustLastInsertID(t, repo, "issue313-forward-a") - if err := repo.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardAID, entryA, 2000).Error; err != nil { + if err := repo.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardAID, entryB2, 2000).Error; err != nil { t.Fatalf("insert forward_port a: %v", err) } @@ -171,7 +170,7 @@ func TestIssue313_EntryPortCrossTunnelConflictContract(t *testing.T) { t.Fatalf("expected update failure due to cross-tunnel port conflict, got success with code 0") } - if !bytes.Contains(res.Body.Bytes(), []byte("端口")) && !bytes.Contains(res.Body.Bytes(), []byte("占用")) { + if !bytes.Contains([]byte(out.Msg), []byte("端口")) && !bytes.Contains([]byte(out.Msg), []byte("占用")) { t.Fatalf("expected port conflict error message, got %q", out.Msg) } diff --git a/go-backend/tests/contract/limiter_sync_failure_contract_test.go b/go-backend/tests/contract/limiter_sync_failure_contract_test.go index 266b66e..555899e 100644 --- a/go-backend/tests/contract/limiter_sync_failure_contract_test.go +++ b/go-backend/tests/contract/limiter_sync_failure_contract_test.go @@ -705,7 +705,7 @@ func TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract(t *testi } oldEntryNodeID := insertNode("issue281-old-entry", "issue281-old-entry-secret", "10.51.0.1", "51000-51010", 0) - newEntryNodeID := insertNode("issue281-new-entry", "issue281-new-entry-secret", "10.51.0.2", "52000-52010", 1) + newEntryNodeID := insertNode("issue281-new-entry", "issue281-new-entry-secret", "10.51.0.2", "51000-51010", 1) exitNodeID := insertNode("issue281-exit", "issue281-exit-secret", "10.51.0.3", "53000-53010", 2) if err := r.DB().Exec(` @@ -881,8 +881,8 @@ func TestTunnelUpdateEntryTransitionsCleanupForwardRuntimeContract(t *testing.T) } entryA := insertNode("issue281-transition-entry-a", "issue281-transition-entry-a-secret", "10.52.0.1", "54000-54010", 0) - entryB := insertNode("issue281-transition-entry-b", "issue281-transition-entry-b-secret", "10.52.0.2", "55000-55010", 1) - entryC := insertNode("issue281-transition-entry-c", "issue281-transition-entry-c-secret", "10.52.0.3", "56000-56010", 2) + entryB := insertNode("issue281-transition-entry-b", "issue281-transition-entry-b-secret", "10.52.0.2", "54000-54010", 1) + entryC := insertNode("issue281-transition-entry-c", "issue281-transition-entry-c-secret", "10.52.0.3", "54000-54010", 2) exitNodeID := insertNode("issue281-transition-exit", "issue281-transition-exit-secret", "10.52.0.4", "57000-57010", 3) if err := r.DB().Exec(` diff --git a/plans/042-sqlite-tunnel-entry-update-deadlock.md b/plans/042-sqlite-tunnel-entry-update-deadlock.md new file mode 100644 index 0000000..3c64c72 --- /dev/null +++ b/plans/042-sqlite-tunnel-entry-update-deadlock.md @@ -0,0 +1,49 @@ +# 042 - SQLite 隧道编辑添加入口节点卡死排查 + +## Issue +- 现象：SQLite 数据库下，编辑已有隧道并新增入口节点时接口卡住；PostgreSQL 下同样操作正常。 +- 初步判断：`tunnelUpdate` 在事务尚未提交时触发了额外 repository 读查询，SQLite 配置 `MaxOpenConns(1)`，容易在同一请求内形成自锁等待。 + +## Goal +- 找出 SQLite 与 PostgreSQL 行为差异的根因。 +- 修复隧道编辑新增入口节点时的阻塞问题，同时不破坏现有的入口端口冲突校验。 +- 补充最小回归测试，锁定“事务内校验不可再次占用根连接”的场景。 + +## Checklist +- [x] 复核 `tunnelUpdate` 在新增入口节点路径上的调用链，确认事务内哪些查询绕过了 `tx`。 +- [x] 为相关 repository 查询补齐 `Tx` 版本，避免 SQLite 单连接下的自锁等待。 +- [x] 调整 handler 中入口端口冲突校验，保证事务内全程复用同一个 `tx`。 +- [x] 增加针对 SQLite 的回归测试，验证事务内校验不会阻塞。 +- [x] 运行相关 handler/backend 测试并记录结果。 + +## Notes +- 重点关注 `validateTunnelEntryPortConflictsForNewEntries`：当前它在 `tx.Commit()` 前执行，但内部调用 `ListForwardsByTunnel` / `ListForwardPorts` / `HasOtherForwardOnNodePort` 等非事务查询。 +- SQLite 在 `go-backend/internal/store/repo/repository.go` 中显式设置了 `SetMaxOpenConns(1)`，因此这种模式在 SQLite 下会比 PostgreSQL 更容易表现为“卡死”。 + +## 实际改动 +- `go-backend/internal/store/repo/repository_control.go` + - 新增 `ListForwardsByTunnelTx`、`ListForwardPortsTx`、`HasOtherForwardOnNodePortTx`，并让原有非事务方法复用统一实现。 +- `go-backend/internal/http/handler/mutations.go` + - `tunnelUpdate` 在事务内执行入口端口冲突校验时显式传入当前 `tx`。 + - `validateTunnelEntryPortConflictsForNewEntries` 改为全程使用事务查询。 + - 新增 `validateForwardPortAvailabilityTx`，避免事务内回落到根连接查询。 +- `go-backend/internal/http/handler/tunnel_entry_sqlite_test.go` + - 新增 SQLite 回归测试，验证开启事务后执行新增入口校验不会阻塞。 + +## 测试结果 + +### 通过 +```bash +cd go-backend && go test ./internal/http/handler/... +``` +- 结果：通过。 + +### 额外检查 +```bash +cd go-backend && go test ./tests/contract/... +``` +- 结果：未全绿；当前失败集中在既有的入口端口语义合同用例： + - `TestIssue313_EntryPortCrossTunnelConflictContract` + - `TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract` + - `TestTunnelUpdateEntryTransitionsCleanupForwardRuntimeContract` +- 备注：这些失败反映的是“新增/切换入口时端口校验预期”与现有合同用例之间的行为差异，不是本次 SQLite 事务自锁修复本身的编译或阻塞问题。 diff --git a/plans/043-entry-contract-semantics-alignment.md b/plans/043-entry-contract-semantics-alignment.md new file mode 100644 index 0000000..ba07a48 --- /dev/null +++ b/plans/043-entry-contract-semantics-alignment.md @@ -0,0 +1,49 @@ +# 043 - Entry Transition Contract Semantics Alignment + +## Issue +- SQLite 阻塞修复完成后，`go test ./tests/contract/...` 暴露出 3 个入口变更相关合同用例失败。 +- 失败原因分成两类： + - Issue 313 用例的数据构造没有真正制造“新增入口节点已被其他转发占用”的冲突。 + - Issue 281 回归用例在后续引入“入口端口必须落在节点端口范围内”后，仍沿用旧的非重叠端口范围数据，和当前产品语义不一致。 + +## Goal +- 对齐这 3 个合同测试与当前后端语义。 +- 保持 SQLite 自锁修复不回退。 +- 让入口节点切换/新增相关合同测试重新稳定通过。 + +## Checklist +- [x] 复核 3 个失败用例的测试数据与当前后端校验语义差异。 +- [x] 调整 Issue 313 用例，确保新增入口节点确实命中“其他转发已占用同节点同端口”。 +- [x] 调整 Issue 281 两个回归用例，使入口切换场景使用与保留端口兼容的节点端口范围。 +- [x] 运行相关合同测试与 handler 测试并记录结果。 + +## Notes +- `validateTunnelEntryPortConflictsForNewEntries` 的占用判定复用 `HasOtherForwardOnNodePort`，语义是“同节点 + 同端口 + 其他转发”，不是全局无节点维度的端口唯一性。 +- 入口切换测试当前更关注 `forward_port` 重建和旧节点运行时清理，因此测试数据应避免被端口范围校验提前拦截。 + +## 实际调整 +- `go-backend/tests/contract/issue313_entry_port_conflict_contract_test.go` + - 将隧道 A 的入口节点改为复用即将添加到隧道 B 的 `entryB2`，确保新增入口时真正命中“同节点同端口已被其他转发占用”。 + - 修正错误消息断言，直接检查 `out.Msg`，避免 JSON 解码后再读 `res.Body` 导致误判。 +- `go-backend/tests/contract/limiter_sync_failure_contract_test.go` + - 将 issue 281 的新入口节点端口范围调整为包含原有保留端口，保持测试关注点在运行时清理/同步，而不是被后续引入的端口范围校验拦截。 + +## 测试结果 + +### 定向回归 +```bash +cd go-backend && go test ./tests/contract/... -run 'TestIssue313_EntryPortCrossTunnelConflictContract|TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract|TestTunnelUpdateEntryTransitionsCleanupForwardRuntimeContract' -v +``` +- 结果：3/3 通过。 + +### Handler +```bash +cd go-backend && go test ./internal/http/handler/... +``` +- 结果：通过。 + +### 全量合同测试 +```bash +cd go-backend && go test ./tests/contract/... +``` +- 结果：通过。