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

@@ -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"

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() {
+	entry := &WalEntry{
-		return errkit.ErrWriteStopped
-	}
-
-	req := ww.queue.Submit([]*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() {
+	entry := &WalEntry{
-		return errkit.ErrWriteStopped
-	}
-
-	req := ww.queue.Submit([]*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 {

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())
+	}
+}