fix: high priority pipelines could be starved by parallel/ label (#8568)

* fix: high priority pipelines could be starved by `parallel/` label

* fix: failed on mysql

Author: Klesh Wong

backend/server/services/pipeline.go

@@ -261,19 +261,32 @@ func dequeuePipeline(runningParallelLabels []string) (pipeline *models.Pipeline,
 	}))
 	// prepare query to find an appropriate pipeline to execute
 	pipeline = &models.Pipeline{}
+	// 1. find out the current highest priority in the queue
+	top_priority := 0
+	var top_priorities []int
+	where_status := dal.Where("status IN ?", []string{models.TASK_CREATED, models.TASK_RERUN, models.TASK_RESUME})
+	err = tx.Pluck("priority", &top_priorities, dal.From(pipeline), where_status, dal.Orderby("priority DESC"), dal.Limit(1))
+	if err != nil {
+		panic(err)
+	}
+	if len(top_priorities) > 0 {
+		top_priority = top_priorities[0]
+	}
+	// 2. pick the earlier runnable pipeline with the highest priority
 	err = tx.First(pipeline,
-		dal.Where("status IN ?", []string{models.TASK_CREATED, models.TASK_RERUN, models.TASK_RESUME}),
+		where_status,
+		dal.Where("priority = ?", top_priority),
 		dal.Join(
 			`left join _devlake_pipeline_labels ON
 				_devlake_pipeline_labels.pipeline_id = _devlake_pipelines.id AND
 				_devlake_pipeline_labels.name LIKE 'parallel/%' AND
 				_devlake_pipeline_labels.name in ?`,
 			runningParallelLabels,
 		),
-		dal.Groupby("priority, id"),
+		dal.Groupby("id"),
 		dal.Having("count(_devlake_pipeline_labels.name)=0"),
 		dal.Select("id"),
-		dal.Orderby("priority DESC, id ASC"),
+		dal.Orderby("id ASC"),
 		dal.Limit(1),
 	)
 	if err == nil {