[Bug]: TaskServiceHolder accepts Create after Close and leaks refresh worker

[Ariznawlll]

问题概述

最新 main 上，taskServiceHolder.Close() 会把 holder 标记为 closed=true，但 Create() 完全不检查这个终态。Close 之后再次收到 CreateTaskService 命令时，holder 会成功创建并保存新的 taskService/refreshableTaskStorage，同时启动 refresh worker。

此后再次调用 holder.Close() 会因为 closed=true 直接返回，不会关闭这个 Close 之后创建的 service，因此后台 worker 和 storage 会永久泄漏。

验证基线

• 仓库：matrixorigin/matrixone
• 分支：最新 main
• Commit：d8ca64575d33455839ddfac2b9b2fe4ea168e003
• Commit 时间：2026-07-17T02:49:42+00:00
• Commit 标题：fix: release RemoteRun stream after peer close (#25791)
• 测试范围：TaskServiceHolder 生命周期；不涉及 2PC

白盒测试过程

新增临时单测 TestTaskServiceHolderRejectsCreateAfterClose：

1. 创建尚未持有 service 的 taskServiceHolder。
2. 调用 holder.Close()，确认返回 nil，此时内部 closed=true。
3. 注入合法的 CreateTaskService 命令。
4. 调用 holder.Get() 检查是否创建并保存了 service。
5. 期望：Create() 返回错误，Get() 返回不存在。
6. 实际：Create() 返回 nil，Get() 能取到新 service。
7. 为避免测试本身泄漏，测试绕过已经失效的 holder Close，直接调用新 service 的 Close()，然后再执行失败断言。

核心流程：

require.NoError(t, holder.Close())

err := holder.Create(validCreateTaskServiceCommand)
service, created := holder.Get()

// 测试清理：holder.Close 已经不会处理这个新 service
if created {
    require.NoError(t, service.Close())
}

require.False(t, err == nil && created,
    "a closed task service holder accepted and retained a newly created service")

执行命令与结果

普通模式，重复 20 次

go test ./pkg/taskservice \
  -run '^TestTaskServiceHolderRejectsCreateAfterClose$' \
  -count=20 -v -timeout=30s

结果：FAIL 20/20。

--- FAIL: TestTaskServiceHolderRejectsCreateAfterClose (0.01s)
Error: Should be false
Messages: a closed task service holder accepted and retained a newly created service

race 模式，重复 10 次

go test -race ./pkg/taskservice \
  -run '^TestTaskServiceHolderRejectsCreateAfterClose$' \
  -count=10 -v -timeout=30s

结果：FAIL 10/10，失败原因一致；没有 race detector 告警。

根因分析

pkg/taskservice/task_service_holder.go：

func (h *taskServiceHolder) Close() error {
    h.mu.Lock()
    defer h.mu.Unlock()

    if h.mu.closed {
        return nil
    }
    h.mu.closed = true
    if h.mu.service == nil {
        return nil
    }
    return h.mu.service.Close()
}

但是 Create() 加锁后只检查 h.mu.service != nil，没有检查 h.mu.closed：

h.mu.Lock()
defer h.mu.Unlock()
if h.mu.service != nil {
    return nil
}

store := newRefreshableTaskStorage(...)
h.mu.service = NewTaskService(h.rt, store)
return nil

newRefreshableTaskStorage 会立即创建 storage 并启动 refreshTask stopper worker。由于 holder 已经 closed，后续 holder.Close() 不会再到达 h.mu.service.Close()。

CN、TN 和 LogService 的 heartbeat 命令处理都会调用 holder Create()：

• pkg/cnservice/server_task.go
• pkg/tnservice/store_task.go
• pkg/logservice/service_task.go

因此生命周期终止与延迟/并发命令之间缺少最终状态保护。

影响

• 已关闭 holder 可以被重新激活，违反 Close 的终态语义。
• 新 refresh worker、TaskStorage 连接及相关资源不会再由 holder 回收。
• Get() 在 holder closed 后仍暴露新 service，调用方可能继续使用逻辑上已关闭的组件。
• 测试、embedded 生命周期或异常命令时序中可能形成 goroutine/连接泄漏。

建议修复

• Create() 获取锁后首先检查 h.mu.closed，closed 时返回明确的 ErrNotReady/invalid-state 错误。
• Get() 可同时检查 closed，避免关闭后继续暴露 service。
• 增加 Close-before-Create 和 Close/Create 并发回归测试。

查重

已搜索 TaskServiceHolder Create after Close leak、taskservice holder closed CreateTaskService、taskServiceHolder closed Create，未发现相同 issue。