flowchart TB
    subgraph Client["客户端层"]
        SDK[Milvus SDK]
        APP[Application]
    end
    
    subgraph Proxy["Proxy层"]
        P1[Proxy-1]
        P2[Proxy-2]
        PN[Proxy-N]
        PCache[GlobalMetaCache<br/>元数据缓存]
    end
    
    subgraph RootCoord["RootCoord核心"]
        API[gRPC API Layer]
        
        subgraph Core["核心组件"]
            Scheduler[Scheduler<br/>DDL调度器]
            Broker[Broker<br/>组件通信层]
            StepExecutor[StepExecutor<br/>步骤执行器]
            DdlLock[DdlTsLockManager<br/>DDL时间戳锁]
        end
        
        subgraph Allocator["资源分配器"]
            TSO[TSOAllocator<br/>时间戳分配]
            IDA[IDAllocator<br/>全局ID分配]
        end
        
        subgraph MetaStore["元数据管理"]
            MetaTable[MetaTable<br/>内存缓存]
            Catalog[Catalog<br/>持久化层]
        end
        
        subgraph Background["后台服务"]
            QuotaCenter[QuotaCenter<br/>配额管理]
            GarbageCollector[GarbageCollector<br/>垃圾回收]
            TimeTickSync[TimeTickSync<br/>时间同步]
        end
    end
    
    subgraph Coordinator["其他Coordinator"]
        DataCoord[DataCoord<br/>数据管理]
        QueryCoord[QueryCoord<br/>查询管理]
    end
    
    subgraph Storage["存储层"]
        ETCD[etcd/TiKV<br/>元数据存储]
        MinIO[MinIO/S3<br/>对象存储]
        MQ[Pulsar/Kafka<br/>消息队列]
    end
    
    %% 请求流程
    SDK --> P1 & P2 & PN
    APP --> P1 & P2 & PN
    P1 & P2 & PN --> API
    
    %% 内部调用链
    API --> Scheduler
    Scheduler --> TSO
    Scheduler --> IDA
    Scheduler --> DdlLock
    Scheduler --> StepExecutor
    
    StepExecutor --> MetaTable
    StepExecutor --> Broker
    
    MetaTable --> Catalog
    Catalog <--> ETCD
    
    Broker --> DataCoord
    Broker --> QueryCoord
    
    %% 后台服务
    QuotaCenter --> P1 & P2 & PN
    GarbageCollector --> DataCoord
    TimeTickSync --> MQ
    TimeTickSync --> TSO
    
    %% DataCoord交互
    DataCoord --> MQ
    DataCoord --> MinIO
    
    %% 缓存失效
    API -.->|InvalidateCache| PCache
    
    style RootCoord fill:#e1f5ff,stroke:#333,stroke-width:3px
    style Core fill:#fff4e6,stroke:#333,stroke-width:2px
    style Allocator fill:#f3e5f5,stroke:#333,stroke-width:2px
    style MetaStore fill:#e8f5e9,stroke:#333,stroke-width:2px
    style Background fill:#fce4ec,stroke:#333,stroke-width:2px

sequenceDiagram
    participant C as Client/Proxy
    participant RC as RootCoord
    participant TSO as TSO Allocator
    participant ETCD as etcd
    
    C->>RC: AllocTimestamp(count=1)
    RC->>TSO: GenerateTSO(1)
    
    alt 缓存足够
        TSO-->>RC: ts=12345678
    else 需要预分配
        TSO->>ETCD: UpdateTSO(+10000)
        ETCD-->>TSO: OK
        TSO->>TSO: 缓存10000个TS
        TSO-->>RC: ts=12345678
    end
    
    RC-->>C: Timestamp

|<--  Physical Time (46 bits) -->|<-- Logical Counter (18 bits) -->|
|   毫秒级物理时间戳                |   逻辑计数器（单位时间内的序号）    |

// 路径：internal/rootcoord/root_coord.go
func (c *Core) CreateCollection(ctx context.Context, in *milvuspb.CreateCollectionRequest) (*commonpb.Status, error) {
    // 1. 健康检查
    if err := merr.CheckHealthy(c.GetStateCode()); err != nil {
        return merr.Status(err), nil
    }
    
    // 2. 创建任务对象
    t := &createCollectionTask{
        baseTask: newBaseTask(ctx, c),
        Req:      in,
    }
    
    // 3. 提交到调度器（关键：串行化）
    if err := c.scheduler.AddTask(t); err != nil {
        return merr.Status(err), nil
    }
    
    // 4. 等待任务完成
    if err := t.WaitToFinish(); err != nil {
        return merr.Status(err), nil
    }
    
    return merr.Success(), nil
}

// 路径：internal/rootcoord/scheduler.go
func (s *scheduler) AddTask(task task) error {
    // 1. 获取锁（保证原子性）
    s.lock.Lock()
    defer s.lock.Unlock()
    
    // 2. 分配任务ID
    if err := s.setID(task); err != nil {
        return err
    }
    
    // 3. 分配时间戳（DDL顺序保证）
    if err := s.setTs(task); err != nil {
        return err
    }
    s.taskHeap.Push(task)
    
    // 4. 将任务放入队列
    s.enqueue(task)
    return nil
}

// 任务循环（串行执行）
func (s *scheduler) taskLoop() {
    for {
        select {
        case task := <-s.taskChan:
            s.execute(task)  // 串行执行，保证顺序性
        }
    }
}

func (s *scheduler) execute(task task) {
    // 1. Prepare阶段：参数校验、资源预分配
    if err := task.Prepare(task.GetCtx()); err != nil {
        task.NotifyDone(err)
        return
    }
    
    // 2. Execute阶段：实际执行
    err := task.Execute(task.GetCtx())
    task.NotifyDone(err)
    
    // 3. 更新最小DDL时间戳
    s.setMinDdlTs()
}

// 路径：internal/rootcoord/create_collection_task.go
func (t *createCollectionTask) Prepare(ctx context.Context) error {
    // 1. 获取Database信息
    db, err := t.core.meta.GetDatabaseByName(ctx, t.Req.GetDbName(), typeutil.MaxTimestamp)
    t.dbID = db.ID
    
    // 2. 参数校验
    if err := t.validate(ctx); err != nil {
        return err
    }
    
    // 3. 解析和校验Schema
    if err := t.prepareSchema(ctx); err != nil {
        return err
    }
    
    // 4. 分配CollectionID
    if err := t.assignCollectionID(); err != nil {
        return err
    }
    
    // 5. 分配PartitionIDs（默认分区 + Partition Key分区）
    if err := t.assignPartitionIDs(ctx); err != nil {
        return err
    }
    
    // 6. 分配VirtualChannels（Shard数量）
    return t.assignChannels()
}

func (t *createCollectionTask) Execute(ctx context.Context) error {
    // 1. 构建Collection模型
    collInfo := model.Collection{
        CollectionID:         t.collID,
        DBID:                 t.dbID,
        Name:                 t.schema.Name,
        Schema:               t.schema,
        VirtualChannelNames:  t.channels.virtualChannels,
        PhysicalChannelNames: t.channels.physicalChannels,
        ShardsNum:            t.Req.ShardsNum,
        State:                pb.CollectionState_CollectionCreating,
        Partitions:           partitions,
        CreateTime:           ts,
    }
    
    // 2. 幂等性检查
    existedColl, err := t.core.meta.GetCollectionByName(ctx, t.Req.GetDbName(), t.Req.GetCollectionName(), typeutil.MaxTimestamp)
    if err == nil {
        // 已存在，检查是否相同
        if !existedColl.Equal(collInfo) {
            return fmt.Errorf("collection already exists with different params")
        }
        return nil  // 幂等返回
    }
    
    // 3. 添加Channel并获取StartPositions
    startPositions, err := t.addChannelsAndGetStartPositions(ctx, ts)
    collInfo.StartPositions = toKeyDataPairs(startPositions)
    
    // 4. 执行Step步骤（事务式）
    return executeCreateCollectionTaskSteps(ctx, t.core, &collInfo, t.Req.GetDbName(), t.dbProperties, ts)
}

func executeCreateCollectionTaskSteps(ctx context.Context, core *Core, col *model.Collection, dbName string, dbProperties []*commonpb.KeyValuePair, ts Timestamp) error {
    undoTask := newBaseUndoTask(core.stepExecutor)
    
    // Step 1: 失效Proxy缓存
    undoTask.AddStep(
        &expireCacheStep{...},      // do: 发送失效通知
        &nullStep{},                // undo: 无操作
    )
    
    // Step 2: 添加Collection元数据
    undoTask.AddStep(
        &addCollectionMetaStep{coll: col},      // do: 写入etcd
        &deleteCollectionMetaStep{collectionID: col.CollectionID},  // undo: 删除元数据
    )
    
    // Step 3: 注册DML Channels
    undoTask.AddStep(
        &nullStep{},                // do: 无操作（已在Execute中完成）
        &removeDmlChannelsStep{pChannels: col.PhysicalChannelNames},  // undo: 移除Channel
    )
    
    // Step 4: 通知DataCoord WatchChannels
    undoTask.AddStep(
        &watchChannelsStep{info: watchInfo{...}},  // do: 调用DataCoord
        &unwatchChannelsStep{...},                  // undo: 取消Watch
    )
    
    // Step 5: 修改Collection状态为Created
    undoTask.AddStep(
        &changeCollectionStateStep{state: CollectionCreated},  // do: 标记为Created
        &nullStep{},                                             // undo: 无操作
    )
    
    // 执行所有步骤（遇错则回滚）
    return undoTask.Execute(ctx)
}

// 路径：internal/rootcoord/undo.go
func (b *baseUndoTask) Execute(ctx context.Context) error {
    for i := 0; i < len(b.todoStep); i++ {
        todoStep := b.todoStep[i]
        
        // 执行正向步骤
        if _, err := todoStep.Execute(ctx); err != nil {
            log.Warn("step failed, starting undo", zap.Error(err))
            
            // 收集已执行的undo步骤
            undoSteps := b.undoStep[:i]
            
            // 异步执行undo（通过StepExecutor）
            go b.stepExecutor.AddSteps(&stepStack{undoSteps})
            return err
        }
    }
    return nil
}

// 路径：internal/rootcoord/broker.go
func (b *ServerBroker) WatchChannels(ctx context.Context, info *watchInfo) error {
    resp, err := b.s.mixCoord.WatchChannels(ctx, &datapb.WatchChannelsRequest{
        CollectionID:    info.collectionID,
        ChannelNames:    info.vChannels,
        StartPositions:  info.startPositions,
        Schema:          info.schema,
        CreateTimestamp: info.ts,
    })
    
    if err != nil || resp.GetStatus().GetErrorCode() != commonpb.ErrorCode_Success {
        return fmt.Errorf("failed to watch channels: %v", err)
    }
    
    return nil
}

sequenceDiagram
    autonumber
    participant Client
    participant Proxy
    participant RC as RootCoord
    participant Scheduler
    participant Task
    participant StepExec as StepExecutor
    participant IDA as IDAllocator
    participant TSO as TSOAllocator
    participant Meta as MetaTable
    participant ETCD as etcd
    participant Broker
    participant DC as DataCoord
    participant MQ as MessageQueue
    
    Client->>Proxy: CreateCollection(name, schema)
    Proxy->>RC: CreateCollection RPC
    
    rect rgb(240, 248, 255)
    Note over RC,Scheduler: API层 - 任务创建与提交
    RC->>RC: 健康检查
    RC->>Task: new createCollectionTask
    RC->>Scheduler: AddTask(task)
    end
    
    rect rgb(255, 250, 240)
    Note over Scheduler,TSO: Scheduler层 - 资源分配
    Scheduler->>IDA: AllocOne()
    IDA->>ETCD: CompareAndSwap(idKey)
    ETCD-->>IDA: OK
    IDA-->>Scheduler: taskID
    
    Scheduler->>TSO: GenerateTSO(1)
    TSO-->>Scheduler: timestamp
    
    Scheduler->>Scheduler: taskQueue.enqueue(task)
    end
    
    rect rgb(243, 229, 245)
    Note over Scheduler,Task: Task层 - Prepare阶段
    Scheduler->>Task: Prepare(ctx)
    Task->>Meta: GetDatabaseByName()
    Meta-->>Task: Database
    
    Task->>Task: validate(schema, shards, limits)
    Task->>Task: prepareSchema(解析Schema)
    
    Task->>IDA: AllocOne()
    IDA-->>Task: collectionID
    
    Task->>IDA: Alloc(partitionCount)
    IDA-->>Task: partitionIDs
    
    Task->>Task: assignChannels(shardsNum)
    Task-->>Scheduler: Prepare完成
    end
    
    rect rgb(232, 245, 233)
    Note over Scheduler,MQ: Task层 - Execute阶段
    Scheduler->>Task: Execute(ctx)
    
    Task->>Task: 构建Collection模型
    Task->>Meta: GetCollectionByName(幂等检查)
    Meta-->>Task: NotFound
    
    Task->>MQ: BroadcastCreateCollectionMsg(获取StartPosition)
    MQ-->>Task: startPositions
    
    Task->>StepExec: executeSteps(undoTask)
    end
    
    rect rgb(255, 228, 225)
    Note over StepExec,DC: StepExecutor层 - 事务式执行
    StepExec->>Proxy: ExpireCacheStep
    Proxy->>Proxy: InvalidateCache
    
    StepExec->>Meta: AddCollectionMetaStep
    Meta->>ETCD: Save(/collections/{id})
    ETCD-->>Meta: OK
    Meta->>Meta: 更新内存缓存
    
    StepExec->>Broker: WatchChannelsStep
    Broker->>DC: WatchChannels(collectionID, vChannels)
    DC->>DC: 创建FlowGraph
    DC->>MQ: Subscribe(vChannels)
    DC-->>Broker: Success
    
    StepExec->>Meta: ChangeCollectionStateStep
    Meta->>ETCD: Update(state=CollectionCreated)
    ETCD-->>Meta: OK
    end
    
    StepExec-->>Task: 所有步骤成功
    Task-->>Scheduler: Execute完成
    Scheduler->>Task: NotifyDone(nil)
    Task-->>RC: WaitToFinish返回
    RC-->>Proxy: Status(Success)
    Proxy-->>Client: CreateCollection成功

// 路径：internal/rootcoord/root_coord.go
func (c *Core) AllocTimestamp(ctx context.Context, in *rootcoordpb.AllocTimestampRequest) (*rootcoordpb.AllocTimestampResponse, error) {
    // 1. 健康检查
    if err := merr.CheckHealthy(c.GetStateCode()); err != nil {
        return &rootcoordpb.AllocTimestampResponse{
            Status: merr.Status(err),
        }, nil
    }
    
    // 2. BlockTimestamp处理（等待物理时间）
    if in.BlockTimestamp > 0 {
        blockTime, _ := tsoutil.ParseTS(in.BlockTimestamp)
        lastTime := c.tsoAllocator.GetLastSavedTime()
        deltaDuration := blockTime.Sub(lastTime)
        if deltaDuration > 0 {
            time.Sleep(deltaDuration + 200*time.Millisecond)
        }
    }
    
    // 3. 生成TSO（核心逻辑）
    ts, err := c.tsoAllocator.GenerateTSO(in.GetCount())
    if err != nil {
        return &rootcoordpb.AllocTimestampResponse{
            Status: merr.Status(err),
        }, nil
    }
    
    // 4. 返回第一个可用时间戳
    ts = ts - uint64(in.GetCount()) + 1
    
    return &rootcoordpb.AllocTimestampResponse{
        Status:    merr.Success(),
        Timestamp: ts,
        Count:     in.GetCount(),
    }, nil
}

// 路径：internal/tso/global_allocator.go
type GlobalTSOAllocator struct {
    mu sync.Mutex
    
    // TSO状态
    lastPhysical int64   // 上次物理时间（毫秒）
    lastLogical  int64   // 上次逻辑计数器
    maxLogical   int64   // 最大逻辑计数器（262144 = 2^18）
    
    // 持久化
    kvBase kv.TxnKV      // etcd/TiKV客户端
    key    string        // TSO存储key
    
    // 更新策略
    updateInterval time.Duration  // 50ms
    saveInterval   time.Duration  // 3s
    lastSaveTime   time.Time
}

func (gta *GlobalTSOAllocator) GenerateTSO(count uint32) (uint64, error) {
    gta.mu.Lock()
    defer gta.mu.Unlock()
    
    // 1. 获取当前物理时间
    physical := time.Now().UnixMilli()
    
    // 2. 物理时间前进，重置逻辑计数器
    if physical > gta.lastPhysical {
        gta.lastPhysical = physical
        gta.lastLogical = 0
    }
    
    // 3. 检查逻辑计数器是否溢出
    if gta.lastLogical+int64(count) >= gta.maxLogical {
        // 等待下一毫秒
        time.Sleep(time.Millisecond)
        gta.lastPhysical = time.Now().UnixMilli()
        gta.lastLogical = 0
    }
    
    // 4. 生成TSO（物理时间 | 逻辑计数器）
    ts := uint64(gta.lastPhysical)<<18 | uint64(gta.lastLogical)
    gta.lastLogical += int64(count)
    
    // 5. 定期持久化到etcd（每3秒）
    if time.Since(gta.lastSaveTime) > gta.saveInterval {
        gta.kvBase.Save(gta.key, gta.lastPhysical)
        gta.lastSaveTime = time.Now()
    }
    
    return ts, nil
}

+-------------------+-------------------+
|  Physical (46位)  | Logical (18位)    |
+-------------------+-------------------+
|  毫秒级时间戳      | 逻辑计数器         |
|  (0-70,368,744,177,663ms) | (0-262,143)  |
+-------------------+-------------------+

示例：
  时间：2024-01-01 00:00:00.000
  Physical: 1704067200000 (毫秒)
  Logical:  12345
  
  TSO计算：
  ts = (1704067200000 << 18) | 12345
     = 446676598505558329
  
  反向解析：
  physical = ts >> 18 = 1704067200000
  logical  = ts & 0x3FFFF = 12345

容量上限：
- 每毫秒最多262,143个TSO
- 理论QPS：262,143,000（2.6亿/秒）
- 实际QPS：>100万/秒（单RootCoord，受锁限制）

持久化策略：
- 每3秒持久化一次physical到etcd
- RootCoord重启后从etcd恢复physical
- 恢复时：physical = saved_physical + 3000ms
- 保证重启后TSO单调递增

特殊情况处理：
1. 逻辑计数器溢出：sleep(1ms)等待下一毫秒
2. 时钟回拨：拒绝分配，返回错误
3. etcd故障：继续分配（最多3秒内的TSO可能丢失）

sequenceDiagram
    autonumber
    participant Proxy
    participant RC as RootCoord
    participant TSO as TSOAllocator
    participant ETCD as etcd
    
    Note over TSO,ETCD: 后台定期持久化
    loop 每3秒
        TSO->>ETCD: Save(tsoKey, lastPhysical)
        Note over TSO: 持久化物理时间<br/>防止重启后时钟回退
    end
    
    rect rgb(240, 248, 255)
    Note over Proxy,RC: 高频TSO分配请求
    Proxy->>RC: AllocTimestamp(count=10)
    RC->>RC: 健康检查
    RC->>TSO: GenerateTSO(10)
    end
    
    rect rgb(255, 250, 240)
    Note over TSO: TSO生成算法（本地）
    TSO->>TSO: 获取当前时间 physical=now()
    
    alt 物理时间前进
        TSO->>TSO: lastPhysical=physical<br/>lastLogical=0
    end
    
    alt 逻辑计数器溢出
        TSO->>TSO: sleep(1ms)等待下一毫秒
        TSO->>TSO: lastPhysical=now()<br/>lastLogical=0
    end
    
    TSO->>TSO: ts = (lastPhysical << 18) | lastLogical
    TSO->>TSO: lastLogical += 10
    end
    
    TSO-->>RC: ts (endTS)
    RC->>RC: firstTS = ts - count + 1
    RC-->>Proxy: Timestamp=firstTS, Count=10
    
    Note over Proxy: 使用范围：[firstTS, firstTS+9]
    
    rect rgb(255, 228, 225)
    Note over Proxy,ETCD: Proxy并发分配多个请求
    par 并发请求
        Proxy->>RC: AllocTimestamp(1)
        RC->>TSO: GenerateTSO(1)
        TSO-->>RC: ts1
    and
        Proxy->>RC: AllocTimestamp(1)
        RC->>TSO: GenerateTSO(1)
        TSO-->>RC: ts2
    and
        Proxy->>RC: AllocTimestamp(1)
        RC->>TSO: GenerateTSO(1)
        TSO-->>RC: ts3
    end
    
    Note over TSO: 所有TS单调递增<br/>ts1 < ts2 < ts3
    end

优化点1：本地生成，避免etcd访问
- 每次请求：0次etcd访问
- P99延迟：<10ms

优化点2：批量分配
- 单次请求可分配多个TS
- 减少RPC次数

优化点3：逻辑计数器
- 单个毫秒内可分配262,143个TS
- 避免频繁获取物理时间

优化点4：定期持久化
- 每3秒持久化1次（vs 每次请求）
- 性能提升：>1000倍

风险：
- RootCoord重启：最多丢失3秒内的TSO信息
- 影响：重启后跳过最多3秒（对MVCC无影响）

flowchart TB
    subgraph API["API入口"]
        A1[CreateCollection]
        A2[DropCollection]
        A3[CreatePartition]
        AN[Other DDL APIs]
    end
    
    subgraph Scheduler["Scheduler调度器"]
        TQ[taskChan<br/>任务队列]
        TH[taskHeap<br/>任务堆]
        TL[taskLoop<br/>串行执行]
        
        subgraph Lock["锁管理"]
            CL[ClusterLock]
            DL[DatabaseLock]
            COL[CollectionLock]
        end
    end
    
    subgraph Allocators["资源分配器"]
        IDA[IDAllocator<br/>任务ID分配]
        TSO[TSOAllocator<br/>时间戳分配]
    end
    
    subgraph Tasks["任务执行"]
        T1[Task.Prepare<br/>参数校验]
        T2[Task.Execute<br/>实际执行]
        T3[Task.NotifyDone<br/>结果通知]
    end
    
    A1 & A2 & A3 & AN --> TQ
    TQ --> TH
    TH --> TL
    
    TL --> IDA
    TL --> TSO
    TL --> Lock
    
    TL --> T1
    T1 --> T2
    T2 --> T3
    
    style Scheduler fill:#e1f5ff,stroke:#333,stroke-width:2px
    style Lock fill:#fff4e6,stroke:#333,stroke-width:2px
    style Tasks fill:#f3e5f5,stroke:#333,stroke-width:2px

// 路径：internal/rootcoord/scheduler.go
type scheduler struct {
    ctx    context.Context
    cancel context.CancelFunc
    wg     sync.WaitGroup
    
    // 资源分配器
    idAllocator  allocator.Interface
    tsoAllocator tso.Allocator
    
    // 任务队列
    taskChan chan task          // 缓冲队列（容量10240）
    taskHeap typeutil.Heap[task]  // 最小堆（按TS排序）
    
    // 并发控制
    lock sync.Mutex
    
    // 最小DDL时间戳
    minDdlTs atomic.Uint64
    
    // 层级锁
    clusterLock    *lock.KeyLock[string]
    databaseLock   *lock.KeyLock[string]
    collectionLock *lock.KeyLock[string]
    lockMapping    map[LockLevel]*lock.KeyLock[string]
}

// 路径：internal/rootcoord/task.go
type task interface {
    GetID() int64
    SetID(id int64)
    GetTs() Timestamp
    SetTs(ts Timestamp)
    GetType() commonpb.MsgType
    
    // 核心方法
    Prepare(ctx context.Context) error
    Execute(ctx context.Context) error
    
    // 异步通知
    WaitToFinish() error
    NotifyDone(err error)
    Notify(err error)
    
    // 锁管理
    GetLockerKey() LockerKey
}

func (s *scheduler) AddTask(task task) error {
    // 1. 判断是否启用锁调度
    if Params.RootCoordCfg.UseLockScheduler.GetAsBool() {
        lockKey := task.GetLockerKey()
        if lockKey != nil {
            return s.executeTaskWithLock(task, lockKey)
        }
    }
    
    // 2. 默认串行化调度
    s.lock.Lock()
    defer s.lock.Unlock()
    
    // 3. 分配任务ID
    if err := s.setID(task); err != nil {
        return err
    }
    
    // 4. 分配时间戳
    if err := s.setTs(task); err != nil {
        return err
    }
    s.taskHeap.Push(task)
    
    // 5. 放入队列
    s.enqueue(task)
    return nil
}

func (s *scheduler) taskLoop() {
    defer s.wg.Done()
    for {
        select {
        case <-s.ctx.Done():
            return
        case task := <-s.taskChan:
            s.execute(task)  // 串行执行
        }
    }
}

func (s *scheduler) execute(task task) {
    defer s.setMinDdlTs()  // 更新最小DDL时间戳
    
    // 1. 记录在队列中的时间
    task.SetInQueueDuration()
    
    // 2. Prepare阶段
    if err := task.Prepare(task.GetCtx()); err != nil {
        task.NotifyDone(err)
        return
    }
    
    // 3. Execute阶段
    err := task.Execute(task.GetCtx())
    
    // 4. 通知完成
    task.NotifyDone(err)
}

func (s *scheduler) executeTaskWithLock(task task, lockerKey LockerKey) error {
    if lockerKey == nil {
        // 叶子节点，执行任务
        if err := s.setID(task); err != nil {
            return err
        }
        s.lock.Lock()
        if err := s.setTs(task); err != nil {
            s.lock.Unlock()
            return err
        }
        s.lock.Unlock()
        s.execute(task)
        return nil
    }
    
    // 获取对应层级的锁
    taskLock := s.lockMapping[lockerKey.Level()]
    if lockerKey.IsWLock() {
        taskLock.Lock(lockerKey.LockKey())
        defer taskLock.Unlock(lockerKey.LockKey())
    } else {
        taskLock.RLock(lockerKey.LockKey())
        defer taskLock.RUnlock(lockerKey.LockKey())
    }
    
    // 递归获取下一层锁
    return s.executeTaskWithLock(task, lockerKey.Next())
}

// 锁层级示例
type LockerKey interface {
    Level() LockLevel      // ClusterLock/DatabaseLock/CollectionLock
    LockKey() string       // 锁的键值
    IsWLock() bool         // 是否写锁
    Next() LockerKey       // 下一层锁
}

// CreateCollection的锁链：
// ClusterLock(RLock) -> DatabaseLock(RLock) -> CollectionLock(WLock)

func (s *scheduler) syncTsLoop() {
    defer s.wg.Done()
    ticker := time.NewTicker(Params.ProxyCfg.TimeTickInterval.GetAsDuration(time.Millisecond))
    defer ticker.Stop()
    
    for {
        select {
        case <-s.ctx.Done():
            return
        case <-ticker.C:
            s.updateLatestTsoAsMinDdlTs()
        }
    }
}

func (s *scheduler) updateLatestTsoAsMinDdlTs() {
    // 创建一个空任务，获取最新TSO
    t := newBaseTask(context.Background(), nil)
    if err := s.AddTask(&t); err != nil {
        log.Warn("failed to update latest ddl ts", zap.Error(err))
    }
}

func (s *scheduler) setMinDdlTs() {
    s.lock.Lock()
    defer s.lock.Unlock()
    
    // 从堆中移除已完成的任务
    for s.taskHeap.Len() > 0 && s.taskHeap.Peek().IsFinished() {
        t := s.taskHeap.Pop()
        s.minDdlTs.Store(t.GetTs())
    }
}

// 用途：TimeTick系统需要知道最小的未完成DDL时间戳
// 保证TimeTick不会超过正在执行的DDL的时间戳

sequenceDiagram
    autonumber
    participant A1 as DDL Task1
    participant A2 as DDL Task2
    participant A3 as DDL Task3
    participant Sch as Scheduler
    participant IDA as IDAllocator
    participant TSO as TSOAllocator
    participant Heap as TaskHeap
    participant Chan as TaskChan
    participant Loop as TaskLoop
    
    rect rgb(240, 248, 255)
    Note over A1,Sch: 并发提交多个DDL任务
    par 并发提交
        A1->>Sch: AddTask(createCollectionTask)
    and
        A2->>Sch: AddTask(dropCollectionTask)
    and
        A3->>Sch: AddTask(createPartitionTask)
    end
    end
    
    rect rgb(255, 250, 240)
    Note over Sch,Heap: 串行化处理（加锁）
    Sch->>Sch: lock.Lock()
    
    Sch->>IDA: AllocOne()
    IDA-->>Sch: taskID=1
    
    Sch->>TSO: GenerateTSO(1)
    TSO-->>Sch: ts=1000
    
    Sch->>Heap: Push(task1, ts=1000)
    Sch->>Chan: taskChan <- task1
    
    Sch->>Sch: lock.Unlock()
    end
    
    rect rgb(243, 229, 245)
    Note over Sch,Heap: 后续任务入队
    Sch->>Sch: lock.Lock()
    
    Sch->>IDA: AllocOne()
    IDA-->>Sch: taskID=2
    
    Sch->>TSO: GenerateTSO(1)
    TSO-->>Sch: ts=1001
    
    Sch->>Heap: Push(task2, ts=1001)
    Sch->>Chan: taskChan <- task2
    
    Sch->>Sch: lock.Unlock()
    end
    
    rect rgb(232, 245, 233)
    Note over Loop,A1: 串行执行Task1
    Loop->>Chan: <-taskChan
    Chan-->>Loop: task1
    
    Loop->>A1: Prepare(ctx)
    A1-->>Loop: nil
    
    Loop->>A1: Execute(ctx)
    A1->>A1: 执行DDL逻辑
    A1-->>Loop: nil
    
    Loop->>A1: NotifyDone(nil)
    A1->>A1: 唤醒WaitToFinish()
    
    Loop->>Heap: Pop已完成任务
    Loop->>Loop: minDdlTs.Store(1000)
    end
    
    rect rgb(255, 228, 225)
    Note over Loop,A2: 串行执行Task2
    Loop->>Chan: <-taskChan
    Chan-->>Loop: task2
    
    Loop->>A2: Prepare(ctx)
    A2-->>Loop: nil
    
    Loop->>A2: Execute(ctx)
    A2->>A2: 执行DDL逻辑
    A2-->>Loop: nil
    
    Loop->>A2: NotifyDone(nil)
    A2->>A2: 唤醒WaitToFinish()
    
    Loop->>Heap: Pop已完成任务
    Loop->>Loop: minDdlTs.Store(1001)
    end
    
    Note over Loop: 串行执行保证DDL顺序性

flowchart TB
    subgraph Task["DDL Task"]
        T1[Task.Execute]
        T2[executeXXXTaskSteps]
    end
    
    subgraph UndoTask["UndoTask构建"]
        UT[newBaseUndoTask]
        
        subgraph Steps["步骤对（Do/Undo）"]
            S1[Step1: ExpireCache / NullStep]
            S2[Step2: AddMeta / DeleteMeta]
            S3[Step3: NullStep / RemoveChannels]
            S4[Step4: WatchChannels / UnwatchChannels]
            S5[Step5: ChangeState / NullStep]
        end
    end
    
    subgraph Executor["StepExecutor"]
        BG[bgStepExecutor<br/>后台执行器]
        SS[stepStack<br/>步骤栈]
        SP[selectPolicy<br/>优先级策略]
    end
    
    subgraph Execution["执行流程"]
        DO[执行Do步骤]
        CHK[检查错误]
        UNDO[触发Undo]
    end
    
    T1 --> T2
    T2 --> UT
    UT --> S1 & S2 & S3 & S4 & S5
    
    S1 & S2 & S3 & S4 & S5 --> DO
    DO --> CHK
    
    CHK -->|成功| S1
    CHK -->|失败| UNDO
    
    UNDO --> BG
    BG --> SS
    SS --> SP
    
    style UndoTask fill:#e1f5ff,stroke:#333,stroke-width:2px
    style Executor fill:#fff4e6,stroke:#333,stroke-width:2px
    style Execution fill:#f3e5f5,stroke:#333,stroke-width:2px

// 路径：internal/rootcoord/undo.go
type baseUndoTask struct {
    todoStep     []nestedStep  // 正向步骤（Do）
    undoStep     []nestedStep  // 回滚步骤（Undo）
    stepExecutor StepExecutor   // 步骤执行器
}

func newBaseUndoTask(stepExecutor StepExecutor) *baseUndoTask {
    return &baseUndoTask{
        todoStep:     make([]nestedStep, 0),
        undoStep:     make([]nestedStep, 0),
        stepExecutor: stepExecutor,
    }
}

// 添加步骤对
func (b *baseUndoTask) AddStep(todoStep, undoStep nestedStep) {
    b.todoStep = append(b.todoStep, todoStep)
    b.undoStep = append(b.undoStep, undoStep)
}

// 路径：internal/rootcoord/step.go
type nestedStep interface {
    Execute(ctx context.Context) ([]nestedStep, error)
    Desc() string          // 步骤描述
    Weight() stepPriority  // 优先级
}

type stepPriority int
const (
    stepPriorityLow       = 0
    stepPriorityNormal    = 1
    stepPriorityImportant = 10
    stepPriorityUrgent    = 1000
)

func (b *baseUndoTask) Execute(ctx context.Context) error {
    if len(b.todoStep) != len(b.undoStep) {
        return errors.New("todo step and undo step length not equal")
    }
    
    for i := 0; i < len(b.todoStep); i++ {
        todoStep := b.todoStep[i]
        
        // 执行正向步骤
        if _, err := todoStep.Execute(ctx); err != nil {
            log.Warn("step failed, starting undo", 
                zap.Error(err), 
                zap.String("desc", todoStep.Desc()))
            
            // 收集已执行步骤的undo
            undoSteps := b.undoStep[:i]
            b.undoStep = nil  // 防止内存泄漏
            
            // 异步执行undo（不阻塞当前调用）
            go b.stepExecutor.AddSteps(&stepStack{undoSteps})
            return err
        }
    }
    return nil
}

// 路径：internal/rootcoord/step_executor.go
type bgStepExecutor struct {
    ctx    context.Context
    cancel context.CancelFunc
    wg     sync.WaitGroup
    
    // 待执行步骤集合
    todoSteps map[*stepStack]struct{}
    mu        sync.Mutex
    
    // 步骤选择策略
    policy selectStepPolicy
    
    // 并发度
    parallel int  // 默认16
}

func (bg *bgStepExecutor) Start() {
    bg.wg.Add(1)
    go bg.scheduleLoop()
}

func (bg *bgStepExecutor) scheduleLoop() {
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-bg.ctx.Done():
            return
        case <-ticker.C:
            bg.process()
        }
    }
}

func (bg *bgStepExecutor) process() {
    bg.mu.Lock()
    // 按优先级选择步骤
    steps := bg.policy(bg.todoSteps)
    bg.mu.Unlock()
    
    // 并发执行
    var wg sync.WaitGroup
    for _, s := range steps {
        wg.Add(1)
        go func(s *stepStack) {
            defer wg.Done()
            
            // 执行步骤栈
            remainSteps := s.Execute(bg.ctx)
            
            bg.mu.Lock()
            defer bg.mu.Unlock()
            
            delete(bg.todoSteps, s)
            if remainSteps != nil {
                // 有剩余步骤，重新加入队列
                bg.todoSteps[remainSteps] = struct{}{}
            }
        }(s)
    }
    wg.Wait()
}

type stepStack struct {
    steps []nestedStep
}

func (s *stepStack) Execute(ctx context.Context) *stepStack {
    steps := s.steps
    for len(steps) > 0 {
        l := len(steps)
        todo := steps[l-1]  // 从后往前执行（栈结构）
        
        childSteps, err := todo.Execute(ctx)
        
        if !retry.IsRecoverable(err) {
            log.Warn("step failed, not recoverable", zap.Error(err))
            return nil
        }
        
        if err != nil {
            log.Warn("step failed, will retry", zap.Error(err))
            return &stepStack{steps: steps}
        }
        
        // 步骤成功，移除并添加子步骤
        steps = steps[:l-1]
        steps = append(steps, childSteps...)
    }
    return nil
}

func (s *stepStack) totalPriority() stepPriority {
    total := stepPriority(0)
    for _, step := range s.steps {
        total += step.Weight()
    }
    return total
}

type addCollectionMetaStep struct {
    baseStep
    coll *model.Collection
}

func (s *addCollectionMetaStep) Execute(ctx context.Context) ([]nestedStep, error) {
    err := s.core.meta.AddCollection(ctx, s.coll)
    return nil, err
}

func (s *addCollectionMetaStep) Desc() string {
    return fmt.Sprintf("add collection to meta table, name: %s, id: %d", 
        s.coll.Name, s.coll.CollectionID)
}

type watchChannelsStep struct {
    baseStep
    info *watchInfo
}

func (s *watchChannelsStep) Execute(ctx context.Context) ([]nestedStep, error) {
    err := s.core.broker.WatchChannels(ctx, s.info)
    return nil, err
}

func (s *watchChannelsStep) Desc() string {
    return fmt.Sprintf("watch channels, ts: %d, collection: %d, vChannels: %v",
        s.info.ts, s.info.collectionID, s.info.vChannels)
}