fix(db): 修复 NextID 与写锁重入死锁
- NextID 改用独立 seqMu,不再依赖数据写锁,杜绝 Lock 内调用 NextID 的死锁 - CaptureAppVolumeForHour 先取 ID 再 Lock - 新增写锁看门狗:长时间占用时转储 goroutine 栈 - 新增 TestNextIDWhileWriteLocked 回归测试
This commit is contained in:
1 parent
f1fc7b4a9a
commit
2e97c04f84
3 files changed
+73
-3
No files matched your search
@@ -6,6 +6,7 @@ import (
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime/pprof"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -24,6 +25,7 @@ type saveRequest struct {
|
||||
// 该包只负责持久化(落盘 + 并发安全),领域实体定义在 models 包。
|
||||
type JsonDB struct {
|
||||
mu sync.RWMutex
|
||||
seqMu sync.Mutex // 独立于 mu:ID 序列不需要与数据锁互相阻塞,避免锁重入死锁
|
||||
dataFile string
|
||||
Users []models.User `json:"users"`
|
||||
Clusters []models.Cluster `json:"clusters"`
|
||||
@@ -873,6 +875,8 @@ func InitDB() error {
|
||||
// 启动序列化保存goroutine
|
||||
d.wg.Add(1)
|
||||
go d.saveWorker()
|
||||
// 写锁看门狗:JSON DB 锁一旦被长时间占用,所有读接口会静默挂起
|
||||
d.StartLockWatchdog(10 * time.Second)
|
||||
|
||||
db = d
|
||||
if migratedLogAccess || migratedSchema {
|
||||
@@ -1068,12 +1072,40 @@ func (d *JsonDB) RequestSaveSync() error {
|
||||
}
|
||||
|
||||
func (d *JsonDB) NextID(table string) uint {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
// 使用独立的 seqMu:调用方即使持有 d.Lock() 也不会死锁(mu 不可重入)。
|
||||
d.seqMu.Lock()
|
||||
defer d.seqMu.Unlock()
|
||||
d.sequences[table]++
|
||||
return d.sequences[table]
|
||||
}
|
||||
|
||||
// StartLockWatchdog 监控写锁是否被长时间占用。
|
||||
//
|
||||
// JSON DB 的写锁是全进程共享的:一旦某个 goroutine 持锁阻塞(网络、磁盘或
|
||||
// 锁重入死锁),所有读库接口(登录、列表、部署)都会静默挂起,而 /health
|
||||
// 不读库仍然正常,表现就是「页面一直转圈、后端像挂了」。
|
||||
// 这里定期 TryLock,连续多次失败就转储 goroutine 栈,让问题在日志里可见。
|
||||
func (d *JsonDB) StartLockWatchdog(interval time.Duration) {
|
||||
go func() {
|
||||
stuck := 0
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
if d.mu.TryLock() {
|
||||
d.mu.Unlock()
|
||||
stuck = 0
|
||||
continue
|
||||
}
|
||||
stuck++
|
||||
if stuck >= 3 {
|
||||
log.Printf("[db] 写锁持续被占用约 %s,转储 goroutine 栈辅助定位", time.Duration(stuck)*interval)
|
||||
_ = pprof.Lookup("goroutine").WriteTo(os.Stderr, 1)
|
||||
stuck = 0
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (d *JsonDB) Lock() {
|
||||
d.mu.Lock()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestNextIDWhileWriteLocked 回归测试:调用方持有写锁时 NextID 不能死锁。
|
||||
//
|
||||
// 历史问题:NextID 内部用 d.mu 加锁,而 CaptureAppVolumeForHour 先 db.Lock()
|
||||
// 再 db.NextID(),Go 的 sync.RWMutex 不可重入,导致该 goroutine 永久持有写锁,
|
||||
// 所有读库接口(登录、列表)全部挂起,而 /health 仍然正常。
|
||||
func TestNextIDWhileWriteLocked(t *testing.T) {
|
||||
db := &JsonDB{sequences: map[string]uint{}}
|
||||
db.Lock()
|
||||
|
||||
done := make(chan uint, 1)
|
||||
go func() { done <- db.NextID("test") }()
|
||||
|
||||
select {
|
||||
case id := <-done:
|
||||
db.Unlock()
|
||||
if id != 1 {
|
||||
t.Fatalf("first id = %d, want 1", id)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
db.Unlock()
|
||||
t.Fatal("NextID deadlocked while write lock is held")
|
||||
}
|
||||
|
||||
// 解锁后继续分配应递增
|
||||
if id := db.NextID("test"); id != 2 {
|
||||
t.Fatalf("second id = %d, want 2", id)
|
||||
}
|
||||
}
|
||||
@@ -147,9 +147,12 @@ func CaptureAppVolumeForHour(src *models.LogSource, hour time.Time) error {
|
||||
if db == nil {
|
||||
return fmt.Errorf("数据库未初始化")
|
||||
}
|
||||
// NextID 内部会自己加锁;必须在 db.Lock() 之前调用。
|
||||
// JsonDB 的 mutex 不可重入,先 Lock 再 NextID 会永久死锁并占住写锁,
|
||||
// 导致所有读库接口(登录、列表)全部卡死。
|
||||
nextID := db.NextID("appVolumeSnapshots")
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
nextID := db.NextID("appVolumeSnapshots")
|
||||
for _, r := range rows {
|
||||
// 幂等:同 (hour, source, ns, app) 已存在则更新,不重复插入
|
||||
found := false
|
||||
|
||||
Reference in new issue
Block a user