别让 Jenkins 决定你的发布节奏：用 Go 自研 DAG 调度引擎的架构决策与踩坑实录

build-A ──┬── deploy-A ──┬── smoke-test-A ──┬── rollout-A
          │              │                   │
build-B ──┤── deploy-B ──┤── smoke-test-B ──┤
          │              │                   │
build-C ─────────────────── deploy-C ────────┘

// Jenkinsfile — 嵌套 parallel 的噩梦
stage('Deploy') {
    parallel {
        stage('Deploy A') { steps { sh 'deploy A' } }
        stage('Deploy B') { steps { sh 'deploy B' } }
    }
}
// C 怎么办？它不依赖 A/B 的 deploy，但不能放在上面的 parallel 里
// 因为 parallel 是"全部完成才继续"
stage('Deploy C') { steps { sh 'deploy C' } }
// 这里的语义已经错了：C 会等 A/B 都 deploy 完才执行

┌─────────────────────────────────────────┐
│         API Layer (Gin + gRPC)          │  ← 流水线定义、触发、状态查询
├─────────────────────────────────────────┤
│      Scheduler Layer (DAG Engine)       │  ← 拓扑排序、分层调度、依赖解析
├─────────────────────────────────────────┤
│      Executor Layer (Worker Pool)       │  ← 并发执行、超时控制、重试
├─────────────────────────────────────────┤
│     State Layer (Redis + MySQL)         │  ← 任务状态持久化、断点续跑
└─────────────────────────────────────────┘

// Pipeline 定义 — 一条完整的发布流水线
type Pipeline struct {
    ID          string
    Name        string
    Tasks       []*Task
    Concurrency int           // 全局并发度上限
    Timeout     time.Duration // 整条流水线超时
    RetryPolicy RetryPolicy   // 全局重试策略
}

// Task 定义 — 流水线中的一个执行单元
type Task struct {
    ID          string
    Name        string
    Type        TaskType     // SHELL, HTTP, K8S_APPLY, GRPC_CALL
    DependsOn   []string     // 依赖的 Task ID 列表
    Command     string       // 执行命令或调用地址
    Timeout     time.Duration
    RetryCount  int
    Conditions  []Condition  // 条件执行（如"仅当上一任务成功时执行"）
    OnFailure   FailureAction // ABORT, SKIP, RETRY, MANUAL
    Idempotent  bool         // 是否幂等（断点续跑时用）
}

// 任务状态机
type TaskState string
const (
    StatePending  TaskState = "PENDING"   // 等待依赖完成
    StateRunning  TaskState = "RUNNING"   // 正在执行
    StateSuccess  TaskState = "SUCCESS"   // 执行成功
    StateFailed   TaskState = "FAILED"    // 执行失败
    StateSkipped  TaskState = "SKIPPED"   // 条件不满足，跳过
    StateAborted  TaskState = "ABORTED"   // 被手动终止
)

type DAG struct {
    nodes    map[string]*Task
    edges    map[string][]string   // 邻接表：from -> [to...]
    inDegree map[string]int
    mu       sync.RWMutex
}

func NewDAG(pipeline *Pipeline) (*DAG, error) {
    dag := &DAG{
        nodes:    make(map[string]*Task),
        edges:    make(map[string][]string),
        inDegree: make(map[string]int),
    }
    // 注册节点
    for _, task := range pipeline.Tasks {
        dag.nodes[task.ID] = task
        dag.inDegree[task.ID] = 0
    }
    // 构建边和入度
    for _, task := range pipeline.Tasks {
        for _, dep := range task.DependsOn {
            if _, ok := dag.nodes[dep]; !ok {
                return nil, fmt.Errorf("task %s depends on unknown task %s", task.ID, dep)
            }
            dag.edges[dep] = append(dag.edges[dep], task.ID)
            dag.inDegree[task.ID]++
        }
    }
    // 环检测（必须在加载阶段完成）
    if cycle := dag.detectCycle(); cycle != nil {
        return nil, fmt.Errorf("cycle detected: %s", formatCycle(cycle))
    }
    return dag, nil
}

// detectCycle — 返回环路径而非简单 true/false
func (d *DAG) detectCycle() []string {
    color := make(map[string]string) // white/gray/black
    for id := range d.nodes { color[id] = "white" }
    var cyclePath []string
    var dfs func(nodeID string, path []string) bool
    dfs = func(nodeID string, path []string) bool {
        color[nodeID] = "gray"
        path = append(path, nodeID)
        for _, neighbor := range d.edges[nodeID] {
            if color[neighbor] == "gray" {
                cyclePath = append(path, neighbor) // 找到环，记录路径
                return true
            }
            if color[neighbor] == "white" {
                if dfs(neighbor, path) { return true }
            }
        }
        color[nodeID] = "black"
        return false
    }
    for id := range d.nodes {
        if color[id] == "white" { if dfs(id, []string{}) { return cyclePath } }
    }
    return nil
}

// GetReadyTasks — 获取当前可执行的任务（入度为 0 且状态为 PENDING）
func (d *DAG) GetReadyTasks() []*Task {
    d.mu.RLock()
    defer d.mu.RUnlock()
    var ready []*Task
    for id, degree := range d.inDegree {
        if degree == 0 {
            if task := d.nodes[id]; task != nil {
                ready = append(ready, task)
            }
        }
    }
    return ready
}

// CompleteTask — 标记任务完成，减少下游入度
func (d *DAG) CompleteTask(taskID string) {
    d.mu.Lock()
    defer d.mu.Unlock()
    for _, downstream := range d.edges[taskID] {
        d.inDegree[downstream]--
    }
}

type WorkerPool struct {
    maxWorkers int
    taskQueue  chan *ExecutableTask
    wg         sync.WaitGroup
    ctx        context.Context
    cancel     context.CancelFunc
}

type TaskResult struct {
    TaskID   string
    State    TaskState
    ExitCode int
    Output   string
    Error    error
    Duration time.Duration
    Retries  int
}

func NewWorkerPool(maxWorkers int, parentCtx context.Context) *WorkerPool {
    ctx, cancel := context.WithCancel(parentCtx)
    return &WorkerPool{
        maxWorkers: maxWorkers,
        taskQueue:  make(chan *ExecutableTask, maxWorkers*2),
        ctx:        ctx,
        cancel:     cancel,
    }
}

func (wp *WorkerPool) Start() {
    for i := 0; i < wp.maxWorkers; i++ {
        wp.wg.Add(1)
        go wp.worker()
    }
}

func (wp *WorkerPool) worker() {
    defer wp.wg.Done()
    for {
        select {
        case <-wp.ctx.Done():
            return
        case execTask := <-wp.taskQueue:
            // recover 防止单个任务 panic 拖垮整个 pool
            result := wp.safeExecute(execTask)
            execTask.Result <- result
        }
    }
}

func (wp *WorkerPool) safeExecute(execTask *ExecutableTask) TaskResult {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("[worker] task %s panicked: %v", execTask.Task.ID, r)
        }
    }()
    return wp.executeWithRetry(execTask)
}

func (wp *WorkerPool) executeWithRetry(execTask *ExecutableTask) TaskResult {
    var lastResult TaskResult
    for attempt := 0; attempt <= execTask.Task.RetryCount; attempt++ {
        ctx, cancel := context.WithTimeout(wp.ctx, execTask.Task.Timeout)
        result := wp.executeSingle(ctx, execTask.Task)
        result.Retries = attempt
        cancel()
        if result.State == StateSuccess { return result }
        lastResult = result
        if attempt < execTask.Task.RetryCount {
            backoff := execTask.Task.RetryDelay * time.Duration(1<<uint(attempt))
            if backoff > 60*time.Second { backoff = 60 * time.Second }
            time.Sleep(backoff)
        }
    }
    return lastResult
}

type Scheduler struct {
    dag        *DAG
    workerPool *WorkerPool
    stateStore StateStore
    eventBus   chan TaskResult
    pipelineID string
}

func (s *Scheduler) Run(ctx context.Context) error {
    // 断点续跑：恢复已完成的任务状态
    if err := s.restoreState(); err != nil {
        return fmt.Errorf("restore state failed: %w", err)
    }
    s.workerPool.Start()
    defer s.workerPool.Stop()

    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case result := <-s.eventBus:
            s.handleResult(result)
        case <-ticker.C:
            s.dispatchReadyTasks()
            if !s.dag.HasRemaining() { return nil }
        }
    }
}

func (s *Scheduler) dispatchReadyTasks() {
    for _, task := range s.dag.GetReadyTasks() {
        s.stateStore.UpdateTaskState(s.pipelineID, task.ID, StateRunning)
        s.dag.markDispatched(task.ID)
        if err := s.workerPool.Submit(&ExecutableTask{Task: task, Result: s.eventBus}); err != nil {
            // 队列满了，回退状态，下次 tick 再调度
            s.stateStore.UpdateTaskState(s.pipelineID, task.ID, StatePending)
            s.dag.unmarkDispatched(task.ID)
        }
    }
}

func (s *Scheduler) handleResult(result TaskResult) {
    s.stateStore.SaveTaskResult(s.pipelineID, result)
    if result.State == StateSuccess || result.State == StateSkipped {
        s.dag.CompleteTask(result.TaskID)
    } else if result.State == StateFailed {
        task := s.dag.nodes[result.TaskID]
        switch task.OnFailure {
        case FailureActionAbort:
            s.abortPipeline(result.TaskID)
        case FailureActionSkip:
            s.dag.CompleteTask(result.TaskID) // 跳过失败任务，继续下游
        case FailureActionManual:
            s.suspendPipeline(result.TaskID, result) // 挂起等人工介入
        }
    }
}

func (s *RedisStateStore) UpdateTaskState(pipelineID, taskID string, state TaskState) error {
    // 1. 先写 Redis（热路径）
    key := fmt.Sprintf("pipeline:%s:task:%s:state", pipelineID, taskID)
    if err := s.redis.Set(ctx, key, string(state), 24*time.Hour).Err(); err != nil {
        return fmt.Errorf("redis write failed: %w", err)
    }
    // 2. 异步写 MySQL（冷路径，失败只记日志不影响调度）
    go func() {
        if _, err := s.mysql.Exec(
            "UPDATE task_executions SET state=?, updated_at=NOW() WHERE pipeline_id=? AND task_id=?",
            string(state), pipelineID, taskID); err != nil {
            log.Printf("[state] mysql write failed for task %s: %v", taskID, err)
        }
    }()
    return nil
}

func (s *Scheduler) restoreState() error {
    completed, _ := s.stateStore.GetCompletedTasks(s.pipelineID)
    for _, result := range completed {
        if result.State == StateSuccess || result.State == StateSkipped {
            s.dag.CompleteTask(result.TaskID)
            s.dag.markDispatched(result.TaskID)
        } else if result.State == StateRunning {
            task := s.dag.nodes[result.TaskID]
            if s.isIdempotent(task) {
                // 幂等任务：回退为 PENDING 重新调度
                s.stateStore.UpdateTaskState(s.pipelineID, result.TaskID, StatePending)
            } else {
                // 非幂等任务：挂起等人工确认
                s.suspendPipeline(result.TaskID, result)
            }
        }
    }
    return nil
}

func (d *DAG) Reverse() *DAG {
    reversed := &DAG{
        nodes:    make(map[string]*Task),
        edges:    make(map[string][]string),
        inDegree: make(map[string]int),
    }
    for id, task := range d.nodes {
        reversed.nodes[id] = &Task{
            ID: task.ID, Name: "rollback-" + task.Name,
            Type: task.Type, Command: task.RollbackCommand,
        }
        reversed.inDegree[id] = 0
    }
    // 反转所有边: 原 from->to 变成 to->from
    for from, tos := range d.edges {
        for _, to := range tos {
            reversed.edges[to] = append(reversed.edges[to], from)
            reversed.inDegree[from]++
        }
    }
    return reversed
}

// 错误写法 — 拓扑排序后线性执行
sorted := topologicalSort(dag)
for _, taskID := range sorted {
    execute(taskID) // 串行！并行变串行！
}

// 正确写法 — 分层并行执行
for {
    ready := dag.GetReadyTasks() // 当前入度为 0 的任务
    if len(ready) == 0 { break }
    for _, task := range ready {
        workerPool.Submit(task) // 并行提交
    }
    waitBatchComplete(ready) // 等这批完成再取下一批
}

// 错误写法 — context 没传到 exec.Command
func (wp *WorkerPool) executeShell(ctx context.Context, cmd string) (TaskState, int, string, error) {
    c := exec.Command("bash", "-c", cmd)
    output, err := c.CombinedOutput() // 不响应 ctx 取消！
    // ...
}

// 正确写法 — 用 CommandContext
func (wp *WorkerPool) executeShell(ctx context.Context, cmd string) (TaskState, int, string, error) {
    c := exec.CommandContext(ctx, "bash", "-c", cmd)
    // ctx 取消时自动发送 SIGKILL
    output, err := c.CombinedOutput()
    // ...
}

# pipeline.yaml — 某物流平台典型发布流水线
name: "物流平台全量发布"
concurrency: 10              # 全局并发度
timeout: 3600s               # 整条流水线超时 1 小时

tasks:
  # 基础设施层
  - id: db-migration
    name: "数据库迁移"
    type: SHELL
    command: "flyway migrate -configFiles=/etc/flyway.conf"
    timeout: 300s
    retry_count: 0            # 数据库迁移不重试
    on_failure: MANUAL        # 失败后人工介入
    idempotent: true

  # 中间件层 — 依赖 db-migration
  - id: deploy-redis-cluster
    name: "部署 Redis 集群"
    type: K8S_APPLY
    command: "kubectl apply -f /manifests/redis/"
    depends_on: [db-migration]
    timeout: 120s
    retry_count: 2
    on_failure: ABORT

  - id: deploy-kafka
    name: "部署 Kafka"
    type: K8S_APPLY
    command: "kubectl apply -f /manifests/kafka/"
    depends_on: [db-migration]
    timeout: 120s
    retry_count: 2
    on_failure: ABORT
    # redis 和 kafka 可以并行（都只依赖 db-migration）

  # 业务层 — 依赖中间件
  - id: deploy-order-service
    name: "部署订单服务"
    type: K8S_APPLY
    command: "kubectl apply -f /manifests/order-service/"
    depends_on: [deploy-redis-cluster, deploy-kafka]
    timeout: 90s

  - id: deploy-delivery-service
    name: "部署配送服务"
    type: K8S_APPLY
    command: "kubectl apply -f /manifests/delivery-service/"
    depends_on: [deploy-redis-cluster, deploy-kafka]
    timeout: 90s
    # order 和 delivery 可以并行（都只依赖中间件层）

  # 接入层 — 依赖所有业务服务
  - id: deploy-gateway
    name: "部署网关"
    type: K8S_APPLY
    command: "kubectl apply -f /manifests/gateway/"
    depends_on: [deploy-order-service, deploy-delivery-service]
    timeout: 60s

  # 验证层
  - id: smoke-test
    name: "冒烟测试"
    type: HTTP
    command: "POST http://smoke-tester.internal/run"
    depends_on: [deploy-gateway]
    timeout: 180s
    conditions:
      - "${deploy-gateway.state} == SUCCESS"

db-migration ──┬── deploy-redis-cluster ──┬── deploy-order-service ──┬── deploy-gateway ── smoke-test
               │                          │                          │
               └── deploy-kafka ──────────┴── deploy-delivery-service ┘

func (s *Scheduler) acquireLeadership(ctx context.Context) bool {
    lockKey := "scheduler:leader"
    ok, err := s.redis.SetNX(ctx, lockKey, s.instanceID, 15*time.Second).Result()
    if err != nil || !ok { return false }
    // 启动续约协程
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()
        for {
            select {
            case <-ctx.Done(): return
            case <-ticker.C: s.redis.Expire(ctx, lockKey, 15*time.Second)
            }
        }
    }()
    return true
}