[Bug]: HAKeeper batch task scheduling discards CN load and repeatedly selects the same CN

[Ariznawlll]

问题概述

最新 main 上，HAKeeper 在同一轮调度多个 Created task 时，每个任务都会重新构造一个过滤后的 CN pool。selectCNs() 没有保留原 pool 中根据 Running task 统计出的负载频次，而分配成功后的频次也只更新临时 pool，不会写回下一任务使用的原 pool。

结果是同一批任务反复选择同一个“旧快照中的最空闲 CN”，形成明显负载倾斜。

验证基线

• 仓库：matrixorigin/matrixone
• 分支：最新 main
• Commit：d8ca64575d33455839ddfac2b9b2fe4ea168e003
• Commit 时间：2026-07-17T02:49:42+00:00
• Commit 标题：fix: release RemoteRun stream after peer close (#25791)
• 测试范围：HAKeeper async task CN 选择与负载均衡；不涉及 2PC

白盒测试过程

新增临时测试 TestScheduleCreatedTasksBalancesBatchAcrossCNs：

1. 集群最初只有 CN a，创建并调度一个任务，使初始负载为 a=1。
2. CN b 加入，此时负载为 a=1, b=0。
3. 同时创建 new-1、new-2、new-3 三个 Created task。
4. 调用一次 scheduler.Schedule，让三个任务在同一调度轮次分配。
5. 正确的 least-loaded 调度应依次更新负载，最终得到 a=2, b=2。
6. 实际三个新任务全部分配给 b，最终为 a=1, b=3。

核心断言：

tasks, err := service.QueryAsyncTask(context.Background(),
    taskservice.WithTaskStatusCond(task.TaskStatus_Running))
assert.NoError(t, err)

counts := map[string]int{}
for _, asyncTask := range tasks {
    counts[asyncTask.TaskRunner]++
}

assert.Equal(t, 2, counts["a"])
assert.Equal(t, 2, counts["b"])

执行命令与结果

普通模式，重复 20 次

go test ./pkg/hakeeper/task \
  -run '^TestScheduleCreatedTasksBalancesBatchAcrossCNs$' \
  -count=20 -v -timeout=30s

结果：FAIL 20/20。每轮日志都显示：

task allocated task-metadata-id=existing task-runner=a
task allocated task-metadata-id=new-1 task-runner=b
task allocated task-metadata-id=new-2 task-runner=b
task allocated task-metadata-id=new-3 task-runner=b

expected: 2
actual  : 1
Messages: batch scheduling did not preserve CN a's running-task load

expected: 2
actual  : 3
Messages: batch scheduling repeatedly selected the same least-loaded snapshot

race 模式，重复 10 次

go test -race ./pkg/hakeeper/task \
  -run '^TestScheduleCreatedTasksBalancesBatchAcrossCNs$' \
  -count=10 -v -timeout=30s

结果：FAIL 10/10，每轮仍为 a=1, b=3；没有 race detector 告警。

根因分析

Schedule() 先通过 Running task 给 workingCNPool 统计负载：

workingCNPool := cnPool.selectCNs(notExpired(...))
expiredTasks := getExpiredTasks(runningTasks, workingCNPool)
s.allocateTasks(createdTasks, workingCNPool)

但是每个任务进入 allocateTask() 后都会再次过滤：

cnPool = cnPool.selectCNs(rules...)
runner := cnPool.min()
// ... Allocate succeeds
heap.Push(cnPool, runner)

这里有两个负载丢失点：

1. cnPool.selectCNs() 创建新 pool 时使用 heap.Push(newPool, cn)。Push 对新 map 从 0 开始递增，因此所有入选 CN 的频次都变成 1，原 pool 的 Running task 频次没有复制。
2. 分配成功后的 heap.Push(cnPool, runner) 只更新本次调用的临时 filtered pool；函数返回后该 pool 被丢弃。下一个 Created task 再从未变化的 workingCNPool 构造相同快照。

即使 task 没有 label/CPU/memory rule，allocateTask 仍然会无条件调用 selectCNs(rules...)，因此普通 async task 也受影响。

影响

• 同一调度周期中的批量任务可能集中到一个 CN，造成 CPU、内存和执行队列热点。
• 已有 Running task 的负载统计在过滤后失效，调度结果不再是 least-loaded。
• task 带 label/resource 约束时同样受影响；候选 CN 越少，热点越明显。
• 调度轮次越大，单轮倾斜越严重，可能影响任务延迟和 CN 稳定性。

建议修复

• selectCNs() 构造新 pool 时复制原始频次，例如使用 newPool.set(cn, p.getFreq(cn.uuid))，不要通过从零累加的 Push 初始化。
• 每次 Allocate 成功后必须更新后续任务共享的基础 pool，而不只是临时 filtered pool。
• 增加多 Created task、已有负载、label/resource 过滤组合的回归测试。

查重

已搜索 HAKeeper task scheduler batch load balance CN、task scheduler selectCNs resets frequency、created tasks same CN unbalanced，未发现相同 issue。