Pipeline Scheduling and Concurrency#

Overview#

DevLake's pipeline scheduler manages how many data-collection pipelines run concurrently and in what order they are dequeued. The core loop lives in RunPipelineInQueue inside backend/server/services/pipeline.go. It combines three complementary mechanisms:

1. Global concurrency cap — PIPELINE_MAX_PARALLEL + a weighted Go semaphore limits the total number of simultaneously running pipelines.
2. Label-based mutual exclusion — pipelines tagged with the same parallel/<group> label are never run at the same time, regardless of available slots.
3. Priority ordering — pipelines carry a numeric Priority field; higher values are scheduled first (added in PR #8534), with a follow-up fix for starvation from parallel/ label interaction (PR #8568).

Concurrency: PIPELINE_MAX_PARALLEL and the Semaphore#

PIPELINE_MAX_PARALLEL is read at startup in pipelineServiceInit(). Its default value is 1 1, meaning pipelines run serially by default. Special cases 2:

• Negative → immediate panic.
• 0 → treated as 10000 (effectively unlimited), with a warning logged.

If the CONSUME_PIPELINES env var is true, the service starts a goroutine running RunPipelineInQueue(pipelineMaxParallel).

Inside RunPipelineInQueue, a semaphore.NewWeighted (from golang.org/x/sync/semaphore) is initialized with pipelineMaxParallel as its capacity. The main loop 3:

1. Acquires one semaphore unit (sema.Acquire(..., 1)) — blocks until a slot is free.
2. Polls dequeuePipeline every second until a runnable pipeline is found.
3. Spawns a goroutine that executes the pipeline and calls sema.Release(1) on completion via defer.

This design means the semaphore limits the number of pipelines in flight, not the number of goroutines waiting.

Label-Based Mutual Exclusion: parallel/ Prefix#

Pipelines can carry arbitrary labels stored in _devlake_pipeline_labels. Labels whose names start with parallel/ are treated as concurrency groups: two pipelines sharing the same parallel/<group> label are never scheduled simultaneously, even if global semaphore slots are available.

How it works in dequeuePipeline:

• The scheduler maintains a runningParallelLabels []string slice — the union of all parallel/-prefixed labels from currently executing pipelines 4.
• The dequeue query does a LEFT JOIN on _devlake_pipeline_labels for rows whose name LIKE 'parallel/%' AND name IN runningParallelLabels, then uses HAVING count(...)=0 to skip any pipeline that shares a label with a running peer 5.
• When a pipeline starts, its parallel/ labels are appended to runningParallelLabels under a mutex 6. They are removed via defer when the pipeline finishes 7.

Labels are persisted to _devlake_pipeline_labels during CreateDbPipeline.

Use case: Assign parallel/project-a to all pipelines for a given project to guarantee they never overlap, while still allowing pipelines for other projects to run concurrently.

Priority-Based Scheduling#

PR #8534 introduced a Priority int field on both the Pipeline and Blueprint models. Higher values mean higher priority. The default is 0.

How priority flows:

• A Blueprint's Priority is copied to NewPipeline.Priority in createPipelineByBlueprint() in blueprint.go 8.
• CreateDbPipeline in pipeline_helper.go persists it to _devlake_pipelines 8.
• A database migration (20250813_add_pipeline_priority.go) adds the priority column 8.

Scheduler change: The dequeuePipeline query's ORDER BY was changed from id ASC to priority DESC, id ASC — within the same priority tier, pipelines are still FIFO by insertion order 8.

Starvation Fix (PR #8568)#

A subtle bug emerged when parallel/ labels interact with priority: if the highest-priority pipeline shares a parallel/ label with an already-running (lower-priority) pipeline, the query would skip it and fall back to a lower-priority pipeline — indefinitely 9.

Fix: dequeuePipeline now performs a two-step query 9:

1. Find top_priority: SELECT MAX(priority) among all queued pipelines (statuses CREATED, RERUN, RESUME).
2. Dequeue within that tier: Add WHERE priority = top_priority to the existing label-exclusion query, ordered only by id ASC.

This ensures the scheduler always considers the highest-priority tier first. If every pipeline in that tier is currently label-blocked, the scheduler waits rather than skipping to a lower-priority tier — preserving strict priority ordering while still respecting label mutex semantics 9.

Key Files and References#