From eb4cb7c655e5bb6796c2514aaa4a714d0b4c3b3e Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 17:28:15 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=86=99=E9=98=9F=E5=88=97=E6=BB=A1?= =?UTF-8?q?=E6=97=B6=20Close=20=E4=B8=8D=E5=86=8D=E4=B8=8E=20enqueue=20?= =?UTF-8?q?=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/DEVIATIONS.md | 10 +++ internal/store/queue.go | 31 +++++--- internal/store/queue_test.go | 134 +++++++++++++++++++++++++++++++++++ 3 files changed, 166 insertions(+), 9 deletions(-) diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 7deb4c5..a85b67c 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -1718,3 +1718,13 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是 - 原因：原先 12 项备注写着未穷尽仍标通过，交付说明写成「通过 23」。 - 备选方案：为每个未测子项补验收用例（本波不做，避免为变绿放松断言）。 - 影响：汇总改为通过 19、部分通过 4（F03/F08/F21/F22）、失败 0。F19 仍引用仓库内 SDK 清单、本波不重跑。 + +### ޸ R3-03 + +1. **дʱ Close enqueue ** + - ڣ2026-09-30 + - ԭissue #67ͣ Drain/Close ڳʱڷأд goroutine ꡣ + - ʵQueue closing źšenqueue ڳ sendMu.RLock ʱ select ͬʱȴ q.chclosing ctx.Done()ͨʱClose Swap(closed) close(closing) ;ͷͷŶд close(q.ch) loop ˳ѹر channel ͡ + - ԭͨʱ Do ռŶڷϣClose дòͣδύ;дҲᶪ + - ѡӸΪ ErrBusyı䱳ѹ壬߷ύ + - Ӱ죺 internal/store/queue.go ԣǨƱš diff --git a/internal/store/queue.go b/internal/store/queue.go index 99ffb1c..20b52e0 100644 --- a/internal/store/queue.go +++ b/internal/store/queue.go @@ -36,10 +36,11 @@ type writeJob struct { type Queue struct { db *sql.DB - ch chan writeJob - done chan struct{} - closed atomic.Bool - sendMu sync.RWMutex + ch chan writeJob + done chan struct{} + closing chan struct{} // Close 时关闭，唤醒持读锁阻塞在发送上的 enqueue + closed atomic.Bool + sendMu sync.RWMutex mu sync.Mutex ready bool @@ -52,11 +53,16 @@ type Queue struct { // NewQueue 创建合并写入队列并启动写 goroutine。 func NewQueue(db *sql.DB) *Queue { + return newQueue(db, queueBuffSize) +} + +func newQueue(db *sql.DB, buffSize int) *Queue { q := &Queue{ - db: db, - ch: make(chan writeJob, queueBuffSize), - done: make(chan struct{}), - ready: true, + db: db, + ch: make(chan writeJob, buffSize), + done: make(chan struct{}), + closing: make(chan struct{}), + ready: true, } go q.loop() return q @@ -115,10 +121,15 @@ func (q *Queue) enqueue(job writeJob) error { return ErrQueueClosed } q.addPending(1) + // 通道满时不得只堵在发送上持有读锁：Close 需要写锁关闭 q.ch。 select { case q.ch <- job: q.sendMu.RUnlock() return nil + case <-q.closing: + q.addPending(-1) + q.sendMu.RUnlock() + return ErrQueueClosed case <-job.ctx.Done(): q.addPending(-1) q.sendMu.RUnlock() @@ -376,11 +387,13 @@ func (q *Queue) Drain(ctx context.Context) error { } // Close 关闭队列：不再接受新任务，并等待写 goroutine 处理完已入队任务后退出。 -// 在写锁内关闭数据通道，避免并发 Do 向已关闭 channel 发送而 panic。 +// 先关闭 closing 唤醒因通道满而阻塞的发送方并释放读锁，再在无发送者时关闭数据通道， +// 避免向已关闭 channel 发送而 panic，也避免与持读锁的 enqueue 死锁。 func (q *Queue) Close() error { if q.closed.Swap(true) { return nil } + close(q.closing) q.sendMu.Lock() close(q.ch) q.sendMu.Unlock() diff --git a/internal/store/queue_test.go b/internal/store/queue_test.go index d51cb26..7f7dede 100644 --- a/internal/store/queue_test.go +++ b/internal/store/queue_test.go @@ -289,3 +289,137 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.upd } } } + +// openSmallQueue 用小缓冲队列替换默认队列，便于测满通道时的 Close/Drain。 +func openSmallQueue(t *testing.T, buf int) *DB { + t.Helper() + dir := t.TempDir() + db, err := Open(dir, "FULL") + if err != nil { + t.Fatal(err) + } + if err := db.Queue.Close(); err != nil { + t.Fatal(err) + } + db.Queue = newQueue(db.Write, buf) + return db +} + +func TestQueueCloseUnblocksFullChannel(t *testing.T) { + t.Parallel() + const buf = 4 + db := openSmallQueue(t, buf) + defer func() { _ = db.Close() }() + q := db.Queue + + hold := make(chan struct{}) + blockerStarted := make(chan struct{}) + blockerErr := make(chan error, 1) + go func() { + blockerErr <- q.Do(context.Background(), func(tx *sql.Tx) error { + close(blockerStarted) + <-hold + return nil + }) + }() + <-blockerStarted + + ctx := context.Background() + var fillWG sync.WaitGroup + for i := 0; i < buf; i++ { + fillWG.Add(1) + go func() { + defer fillWG.Done() + _ = q.Do(ctx, func(tx *sql.Tx) error { return nil }) + }() + } + deadline := time.Now().Add(2 * time.Second) + for q.Len() < buf+1 && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + if q.Len() < buf+1 { + t.Fatalf("channel not full: len=%d", q.Len()) + } + + blockedErr := make(chan error, 1) + go func() { + blockedErr <- q.Do(ctx, func(tx *sql.Tx) error { return nil }) + }() + // 等额外 Do 堵在 enqueue（pending 超过通道容量+正在执行的一条）。 + deadline = time.Now().Add(2 * time.Second) + for q.Len() < buf+2 && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + + closeDone := make(chan error, 1) + go func() { closeDone <- q.Close() }() + + select { + case err := <-blockedErr: + if !errors.Is(err, ErrQueueClosed) { + t.Fatalf("blocked Do: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Close did not unblock full-channel enqueue") + } + + close(hold) + select { + case err := <-closeDone: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("Close hung after writer released") + } + <-blockerErr + fillWG.Wait() +} + +func TestQueueDrainTimeoutWhileWriterBlocked(t *testing.T) { + t.Parallel() + const buf = 4 + db := openSmallQueue(t, buf) + defer func() { _ = db.Close() }() + q := db.Queue + + hold := make(chan struct{}) + started := make(chan struct{}) + go func() { + _ = q.Do(context.Background(), func(tx *sql.Tx) error { + close(started) + <-hold + return nil + }) + }() + <-started + + ctx := context.Background() + var wg sync.WaitGroup + for i := 0; i < buf; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _ = q.Do(ctx, func(tx *sql.Tx) error { return nil }) + }() + } + deadline := time.Now().Add(2 * time.Second) + for q.Len() < buf+1 && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + + drainCtx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond) + defer cancel() + start := time.Now() + err := q.Drain(drainCtx) + elapsed := time.Since(start) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Drain err=%v want deadline", err) + } + if elapsed > 500*time.Millisecond { + t.Fatalf("Drain took %s, should return on timeout", elapsed) + } + + close(hold) + wg.Wait() +}