feat: 流水线队列引擎 — max_executors并发控制 + Jenkins风格构建队列+取消

backend/internal/api/handlers/pipeline_executor.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"sync"
 	"time"
 
 	"ezops/internal/models"
@@ -108,6 +109,7 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P
 		log("success", "完成", "流水线执行成功 (0s)")
 		updateStatus("success", 0)
 		updatePipelineStats("success")
+		h.dequeueNext(pipeline.ID)
 		h.flushLogsToDB(ctx, runID, logKey)
 		return
 	}
@@ -144,6 +146,7 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P
 			log("warn", stage.Name, "流水线已被停止")
 			updateStatus("cancelled", int(time.Since(totalStart).Seconds()))
 			updatePipelineStats("cancelled")
+			h.dequeueNext(pipeline.ID)
 			h.flushLogsToDB(ctx, runID, logKey)
 			return
 		}
@@ -174,6 +177,7 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P
 			log("error", stage.Name, fmt.Sprintf("✗ %s 失败 (%v): %v", stage.Name, stageDur.Round(time.Second), err))
 			updateStatus("failed", int(time.Since(totalStart).Seconds()))
 			updatePipelineStats("failed")
+			h.dequeueNext(pipeline.ID)
 			h.flushLogsToDB(ctx, runID, logKey)
 			return
 		}
@@ -182,10 +186,10 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P
 
 	totalDur := int(time.Since(totalStart).Seconds())
 	updateStatus("success", totalDur)
-	updatePipelineStats("success")
 	log("success", "完成", fmt.Sprintf("流水线执行成功 (%ds)", totalDur))
 
-	// 执行结束 → Redis 日志批量写入 MySQL
+	updatePipelineStats("success")
+	h.dequeueNext(pipeline.ID)
 	h.flushLogsToDB(ctx, runID, logKey)
 }
 
@@ -549,6 +553,93 @@ func cleanOldWorkspaces(baseDir string, maxAge time.Duration) {
 	}
 }
 
+// ==================== 流水线队列引擎（Jenkins 风格并发控制）====================
+
+var (
+	pipelineRunningCount = make(map[uint]int)    // pipelineID → 当前运行数
+	pipelineQueues       = make(map[uint][]uint)  // pipelineID → 排队 runID 列表
+	queueMu              sync.Mutex
+)
+
+// tryExecuteOrQueue 触发流水线时检查并发限制
+// 如果当前运行数 < maxExecutors → 立即执行
+// 否则 → 加入队列 (status=queued)
+func (h *PipelineHandler) tryExecuteOrQueue(pipeline models.Pipeline, run models.PipelineRun) {
+	max := pipeline.MaxExecutors
+	if max <= 0 { max = 1 }
+
+	queueMu.Lock()
+	running := pipelineRunningCount[pipeline.ID]
+	if running < max {
+		pipelineRunningCount[pipeline.ID] = running + 1
+		queueMu.Unlock()
+		// 立即执行
+		h.db.Model(&models.PipelineRun{}).Where("id = ?", run.ID).Update("status", "running")
+		go h.executePipeline(pipeline, run)
+		return
+	}
+	// 加入队列
+	pipelineQueues[pipeline.ID] = append(pipelineQueues[pipeline.ID], run.ID)
+	queueMu.Unlock()
+	h.db.Model(&models.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued")
+}
+
+// dequeueNext 流水线运行结束后，从队列中取下一个执行
+func (h *PipelineHandler) dequeueNext(pipelineID uint) {
+	queueMu.Lock()
+	running := pipelineRunningCount[pipelineID]
+	if running > 0 {
+		pipelineRunningCount[pipelineID] = running - 1
+	}
+
+	queue := pipelineQueues[pipelineID]
+	if len(queue) == 0 {
+		queueMu.Unlock()
+		return
+	}
+	nextRunID := queue[0]
+	pipelineQueues[pipelineID] = queue[1:]
+	pipelineRunningCount[pipelineID] = pipelineRunningCount[pipelineID] + 1
+	queueMu.Unlock()
+
+	// 加载 pipeline + run
+	var pipeline models.Pipeline
+	if err := h.db.First(&pipeline, pipelineID).Error; err != nil { return }
+	var run models.PipelineRun
+	if err := h.db.First(&run, nextRunID).Error; err != nil { return }
+
+	h.db.Model(&run).Update("status", "running")
+	go h.executePipeline(pipeline, run)
+}
+
+// GetQueueStatus 查询流水线队列状态（供 API 使用）
+func (h *PipelineHandler) GetQueueStatus(pipelineID uint) map[string]interface{} {
+	queueMu.Lock()
+	defer queueMu.Unlock()
+	running := pipelineRunningCount[pipelineID]
+	queued := pipelineQueues[pipelineID]
+	return map[string]interface{}{
+		"running": running,
+		"queued":  len(queued),
+		"queue_ids": queued,
+	}
+}
+
+// CancelQueued 取消排队中的构建
+func (h *PipelineHandler) CancelQueued(pipelineID uint, runID uint) error {
+	queueMu.Lock()
+	defer queueMu.Unlock()
+	queue := pipelineQueues[pipelineID]
+	for i, id := range queue {
+		if id == runID {
+			pipelineQueues[pipelineID] = append(queue[:i], queue[i+1:]...)
+			h.db.Model(&models.PipelineRun{}).Where("id = ?", runID).Update("status", "cancelled")
+			return nil
+		}
+	}
+	return fmt.Errorf("run %d not in queue", runID)
+}
+
 func findFiles(dir string, exts ...string) []string {
 	var result []string
 	entries, err := os.ReadDir(dir)