From 408138b3c808aef4a019b811be39e5859698849b Mon Sep 17 00:00:00 2001 From: dailz Date: Thu, 18 Jun 2026 13:25:30 +0800 Subject: [PATCH] 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 construct entry outside the lock, then 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. Run with -count=50 for regression detection. Verified: all existing tests pass. New concurrent tests pass with -count=50 (0 panics). go test -race ./... green. Audit context: docs/audit-3.2.md C7. --- .omo/plans/fix-c7-put-close-race.md | 339 ++++++++++++++++++++++++++++ wal/writer.go | 43 +++- wal/writer_test.go | 99 ++++++++ 3 files changed, 469 insertions(+), 12 deletions(-) create mode 100644 .omo/plans/fix-c7-put-close-race.md diff --git a/.omo/plans/fix-c7-put-close-race.md b/.omo/plans/fix-c7-put-close-race.md new file mode 100644 index 0000000..3d6f100 --- /dev/null +++ b/.omo/plans/fix-c7-put-close-race.md @@ -0,0 +1,339 @@ +# 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" diff --git a/wal/writer.go b/wal/writer.go index ad38966..4b4ef9e 100644 --- a/wal/writer.go +++ b/wal/writer.go @@ -58,6 +58,7 @@ type WalWriter struct { seqManager *SequenceManager memTables *MemTableList writeStopped atomic.Bool + submitMu sync.RWMutex // C7 fix: protects Submit vs Close critical section done chan struct{} wg sync.WaitGroup closeOnce sync.Once @@ -311,31 +312,43 @@ func (ww *WalWriter) sendSuccess(requests []*CommitRequest, baseSequence uint64, // Put stores key with value. func (ww *WalWriter) Put(key, value []byte) error { - if ww.writeStopped.Load() { - return errkit.ErrWriteStopped - } - - req := ww.queue.Submit([]*WalEntry{{ + entry := &WalEntry{ OpType: OpPut, ValueKind: VKInline, Key: cloneBytes(key), Value: cloneBytes(value), - }}) + } + + // C7 fix: RLock protects check + Submit. Close holds write lock during + // writeStopped.Store + queue.Close, so Submit cannot send on a 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 } // Delete removes key. func (ww *WalWriter) Delete(key []byte) error { - if ww.writeStopped.Load() { - return errkit.ErrWriteStopped - } - - req := ww.queue.Submit([]*WalEntry{{ + entry := &WalEntry{ OpType: OpDelete, ValueKind: VKNone, Key: cloneBytes(key), - }}) + } + + 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 } @@ -368,8 +381,14 @@ func (ww *WalWriter) IsWriteStopped() bool { // Close drains queued writes, stops the writer goroutine, syncs, and closes the segment. func (ww *WalWriter) Close() error { ww.closeOnce.Do(func() { + // C7 fix: write lock waits for all in-flight Put/Delete Submits to + // complete. While we hold this lock, no Submit can be in progress, + // so queue.Close is safe (no send-on-closed-channel panic). + 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 { diff --git a/wal/writer_test.go b/wal/writer_test.go index 023dc42..6174593 100644 --- a/wal/writer_test.go +++ b/wal/writer_test.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "github.com/dailz/go-kv/config" @@ -164,3 +165,101 @@ func TestWriteStoppedAfterIOError(t *testing.T) { _ = ww.Close() } + +// -------- C7 regression guards -------- + +// 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. +// +// Note: this test guarantees "no panic AFTER fix", not "must panic BEFORE +// fix" — the race window is narrow (Put check → Submit, a few instructions) +// and the default queue capacity (10000) means sends rarely block. Run +// with -count=50 to increase regression detection probability. +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()) + } +} + +// Regression guard for C7: same as above but for Delete. +func TestConcurrentDeleteCloseNoPanic(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.Delete([]byte("k")) + }() + } + + wg.Add(1) + go func() { + defer wg.Done() + <-start + _ = ww.Close() + }() + + close(start) + wg.Wait() + + if panicCount.Load() > 0 { + t.Fatalf("concurrent Delete + Close caused %d panic(s) — C7 regression", + panicCount.Load()) + } +}