TimingWheel has multiple logic issues, with closure capture and missing Wait being the most severe!

[jeryaiwei]

go-zero/core/collection/timingwheel.go

Lines 165 to 180 in 0a72444

	func (tw *TimingWheel) drainAll(fn func(key, value any)) {
	runner := threading.NewTaskRunner(drainWorkers)
	for _, slot := range tw.slots {
	for e := slot.Front(); e != nil; {
	task := e.Value.(*timingEntry)
	next := e.Next()
	slot.Remove(e)
	e = next
	if !task.removed {
	runner.Schedule(func() {
	fn(task.key, task.value)
	})
	}
	}
	}
	}

// 1. 修复 drainAll
func (tw *TimingWheel) drainAll(fn func(key, value any)) {
   runner := threading.NewTaskRunner(drainWorkers)
   for _, slot := range tw.slots {
       for e := slot.Front(); e != nil; {
           task := e.Value.(*timingEntry)
           next := e.Next()
           slot.Remove(e)
           e = next
           if !task.removed {
               k, v := task.key, task.value  // ✓ 复制变量
               runner.Schedule(func() {
                   fn(k, v)
               })
           }
           // 🔥 建议：同时清理 timers 中的数据
           tw.timers.Del(task.key)
       }
   }
   runner.Wait()  // ✓ 等待完成
}

go-zero/core/collection/timingwheel.go

Lines 266 to 278 in 0a72444

	func (tw *TimingWheel) runTasks(tasks []timingTask) {
	if len(tasks) == 0 {
	return
	}
	go func() {
	for i := range tasks {
	threading.RunSafe(func() {
	tw.execute(tasks[i].key, tasks[i].value)
	})
	}
	}()
	}

// 2. 修复 runTasks
func (tw *TimingWheel) runTasks(tasks []timingTask) {
    if len(tasks) == 0 {
        return
    }

    go func() {
        for i := range tasks {
            task := tasks[i]  // ✓ 复制
            threading.RunSafe(func() {
                tw.execute(task.key, task.value)
            })
        }
    }()
}

[kevwan]

1. drainAll - Closure Capture

Status: ✅ Not a bug, but improvement recommended

The current code is actually safe because task is declared inside the loop body:

for e := slot.Front(); e != nil; {
    task := e.Value.(*timingEntry)  // ← New variable each iteration
    // ...
}

In Go, variables declared inside the loop body (not in the for statement) are created anew for each iteration, so each closure captures a different variable instance.

However, explicitly copying the values as suggested would improve code clarity and robustness.

Missing runner.Wait() - ❌ This IS a real bug!
The function returns immediately without waiting for scheduled tasks to complete, which can cause race conditions and unexpected behavior.

2. runTasks - Loop Variable Capture

Status: ✅ Not a bug in current implementation

The closure captures i, but threading.RunSafe executes synchronously:

func RunSafe(fn func()) {
    defer rescue.Recover()
    fn()  // ← Immediate execution
}

So tasks[i] is accessed with the correct value of i before the next iteration.

However, copying the value as suggested is still good defensive programming and follows Go best practices.