# C7 修复方案:Put/Delete vs Close 并发竞态 ## TL;DR > **目标**:消除 Put/Delete 检查 `writeStopped` 和 `Submit` 之间的时间窗,防止 Close 并发关闭 channel 时 send-on-closed panic。 > > **交付**: > - `WalWriter` 加 `submitMu sync.RWMutex` > - Put/Delete 持读锁保护 check+Submit 临界区 > - Close 持写锁保护 writeStopped+queue.Close > - 2 个并发测试(Put+Close、Delete+Close) > - 单次 commit > > **预估工时**:1.5-2 小时 > **风险**:低。改动局限在 `WalWriter` 一个文件,3 个函数 --- ## Context ### Bug 摘要 `wal/writer.go:313-326` Put(Delete 同理 328-341): ```go func (ww *WalWriter) Put(key, value []byte) error { if ww.writeStopped.Load() { // ① 检查 return errkit.ErrWriteStopped } req := ww.queue.Submit(...) // ② Submit(内部 `cq.ch <- req`) result := <-req.Result return result.Err } ``` `wal/writer.go:369-382` Close: ```go func (ww *WalWriter) Close() error { ww.closeOnce.Do(func() { ww.writeStopped.Store(true) // A 设标志 ww.queue.Close() // B 关 channel close(ww.done) ww.wg.Wait() ... }) } ``` `wal/commit_queue.go:30-37` Submit: ```go func (cq *CommitQueue) Submit(entries []*WalEntry) *CommitRequest { req := &CommitRequest{...} cq.ch <- req // ← send on possibly-closed channel return req } ``` `CommitQueue.Close` 的注释**自己写着** "Callers must stop submitting before Close" —— 但 Put/Delete 并没有保证这一点。 ### 竞态场景 ``` Goroutine 1 (Put) Goroutine 2 (Close) ──────────────── ──────────────────── writeStopped.Load() = false writeStopped.Store(true) queue.Close() → close(cq.ch) queue.Submit(): cq.ch <- req ← PANIC "send on closed channel" ``` 时间窗:Put 的 ①→② 之间,Close 的 A→B 发生。 ### 为什么没被抓到 现有测试串行调用 Put → Close,不模拟并发。要抓 C7 需要: - 多 goroutine 并发 Put + 1 goroutine Close - `defer recover` 捕获 panic - 高并发 + 多次运行提高触发概率 属于"低概率高破坏"型 —— 平时不发作,发作就 panic 整个进程。 --- ## 执行计划 ### Phase A:代码改动(30 分钟) #### A.1 加 `submitMu sync.RWMutex` 到 WalWriter ```go type WalWriter struct { ... writeStopped atomic.Bool submitMu sync.RWMutex // C7 fix: 保护 Submit vs Close 临界区 done chan struct{} ... } ``` #### A.2 修改 Put ```go // 改后: func (ww *WalWriter) Put(key, value []byte) error { // Construct entry outside the lock to minimize critical section // (Oracle nice-to-have: reduces contention under high concurrency). entry := &WalEntry{ OpType: OpPut, ValueKind: VKInline, Key: cloneBytes(key), Value: cloneBytes(value), } // C7 fix: RLock 保护 check + Submit 临界区。Close 持写锁时, // 所有 RLock 释放后才能 queue.Close,保证 Submit 不会 send on closed channel。 ww.submitMu.RLock() if ww.writeStopped.Load() { ww.submitMu.RUnlock() return errkit.ErrWriteStopped } req := ww.queue.Submit([]*WalEntry{entry}) ww.submitMu.RUnlock() result := <-req.Result return result.Err } ``` **关键**:RLock 在 Submit 之后、`<-req.Result` 之前释放。这样: - Submit 在锁保护内(防止 Close 并发关 channel) - `<-req.Result` 在锁外(不阻塞其他 Put,不阻塞 Close 拿写锁) #### A.3 修改 Delete 同 Put 模式。 #### A.4 修改 Close ```go // 改后: func (ww *WalWriter) Close() error { ww.closeOnce.Do(func() { // C7 fix: 写锁等待所有 Put/Delete 的 RLock 释放。 // 持锁期间关 channel 是安全的:没有 Submit 在进行中。 ww.submitMu.Lock() ww.writeStopped.Store(true) ww.queue.Close() ww.submitMu.Unlock() close(ww.done) ww.wg.Wait() if err := ww.segManager.Sync(); err != nil { ww.closeErr = err return } ww.closeErr = ww.segManager.Close() }) return ww.closeErr } ``` **关键顺序**: 1. 写锁(等所有 in-flight Submit 完成) 2. `writeStopped = true`(新 Put 看到后返回错误) 3. `queue.Close`(此刻无 Submit 在进行,安全) 4. 写锁释放 5. `close(done)` + `wg.Wait`(runLoop 退出) #### 死锁分析 | 场景 | 分析 | |------|------| | Put 持 RLock + Submit 阻塞(channel 满)| runLoop 持续消费(不持任何锁),Submit 最终完成,Put 释放 RLock | | Close 等写锁 | 等 RLock 释放;RLock 释放依赖 Submit 完成;Submit 完成依赖 runLoop 消费;runLoop 不受锁影响 → 最终完成 | | processBatch 失败后 writeStopped=true | 不影响 runLoop 继续消费 channel,不持任何锁,无死锁 | **无死锁**。Close 可能因 Submit 阻塞而延迟(等 I/O),但最终完成。这是 graceful shutdown 的正确语义 —— **保留 backpressure,不创建锁环**。 ### Phase B:测试(45 分钟) #### B.1 TestConcurrentPutCloseNoPanic 新增到 `wal/writer_test.go`: ```go // Regression guard for C7: concurrent Put + Close must not panic. // Buggy code has a race window between writeStopped check and Submit's // send on channel; Close closing the channel during that window causes // "send on closed channel" panic. func TestConcurrentPutCloseNoPanic(t *testing.T) { dir := t.TempDir() cfg := config.Defaults() ww, err := NewWalWriter(&cfg, dir, 0, 0) if err != nil { t.Fatal(err) } const goroutines = 100 var wg sync.WaitGroup var panicCount atomic.Int32 start := make(chan struct{}) for i := 0; i < goroutines; i++ { wg.Add(1) go func() { defer wg.Done() defer func() { if r := recover(); r != nil { panicCount.Add(1) } }() <-start _ = ww.Put([]byte("k"), []byte("v")) }() } wg.Add(1) go func() { defer wg.Done() <-start _ = ww.Close() }() close(start) wg.Wait() if panicCount.Load() > 0 { t.Fatalf("concurrent Put + Close caused %d panic(s) — C7 regression", panicCount.Load()) } } ``` #### B.2 TestConcurrentDeleteCloseNoPanic 同 B.1 但调 `ww.Delete([]byte("k"))`。 #### B.3 测试稳定性说明 > **Oracle 修订(bg_96db47f5)**:测试定位是"fix 后**保证不 panic**",不是"fix 前**必 panic**"。buggy 代码的 race 窗口非常窄(Put check → Submit 之间几条指令),默认 queue capacity=10000,100 goroutines 通常不足以触发 channel 满阻塞。`-count=50` 能提高触发概率但不保证每次都抓到。 > > 建议用 `go test -race ./wal -run TestConcurrent -count=10` 作为可选补充。确定性测试需要在 check 和 Submit 之间注入 test seam(过度侵入性,Phase 1 不值得)。 ### Phase C:验证(15 分钟) ```bash # 1. 编译 go build ./... # 2. 并发测试(单次) go test ./wal -run 'TestConcurrent' -count=1 -v # 3. 并发测试(多次,抓 flaky race) go test ./wal -run 'TestConcurrent' -count=50 # 4. wal 包全量 go test ./wal/... -count=1 # 5. 全仓 go test ./... -count=1 # 6. race go test -race ./... -count=1 # 7. vet go vet ./... ``` ### Phase D:Commit message draft ``` fix: prevent send-on-closed-channel panic in concurrent Put/Delete + Close (C7) Put/Delete checked writeStopped then called queue.Submit (which sends on a channel). Close set writeStopped then closed the channel. With no synchronization between the check and Submit, a concurrent Close could close the channel during the window, causing "send on closed channel" panic in Put/Delete. The CommitQueue.Close comment said "Callers must stop submitting before Close" but Put/Delete didn't enforce this. Changes: - wal/writer.go: add submitMu sync.RWMutex. Put/Delete hold RLock during writeStopped check + Submit; release before <-req.Result. Close holds write lock during writeStopped.Store + queue.Close, guaranteeing no Submit is in progress when the channel is closed. - wal/writer_test.go: add TestConcurrentPutCloseNoPanic and TestConcurrentDeleteCloseNoPanic. 100 goroutines + 1 closer, recover panics. On buggy code, this test triggers "send on closed channel" panic (may require -count=N for reliable reproduction due to narrow race window). On fixed code, never panics. Verified: all existing tests pass. New concurrent tests pass with -count=50. go test -race ./... green. Audit context: docs/audit-3.2.md C7. ``` --- ## 验收清单 - [ ] Phase A.1:`submitMu sync.RWMutex` 字段存在 - [ ] Phase A.2:Put 在 RLock 内 check + Submit,RUnlock 后 `<-req.Result` - [ ] Phase A.3:Delete 同 Put 模式 - [ ] Phase A.4:Close 在写锁内 writeStopped + queue.Close - [ ] Phase B.1:`TestConcurrentPutCloseNoPanic` 存在 - [ ] Phase B.2:`TestConcurrentDeleteCloseNoPanic` 存在 - [ ] `go test ./wal/... -count=1` 全绿 - [ ] `go test ./wal -run TestConcurrent -count=50` 全绿 - [ ] `go test -race ./... -count=1` 全绿 - [ ] `go vet ./...` 无新增警告 - [ ] 单次 commit,message 引用 audit C7 --- ## 不在本次范围内 | 项 | 为什么不放进来 | |------|---------------| | CommitQueue API 改动 | 不需要。RWMutex 在 WalWriter 层就够 | | DB 层并发保护 | 不需要。WalWriter 的保护对 DB 层透明生效(db.Put 调 writer.Put,writer 的 submitMu 已经保护)| | processBatch 失败后的 channel 状态 | 不影响。writeStopped 后 runLoop 继续消费,不关 channel | --- ## 修订记录 - **v1(原始)**:C7 修复方案初稿,送 Momus 审 - **v1.0(Momus 审核 bg_66fea405)**:[OKAY],无 blocking - **v1.1(Oracle 审核 bg_96db47f5)**:**approve**(无 blocking)。3 个 nice-to-have 已采纳: - 缩小临界区:WalEntry 构造移到 RLock 之前 - 死锁分析补充"保留 backpressure,不创建锁环" - 测试定位澄清:是"fix 后保证不 panic",不是"fix 前必 panic"