Workflow Dispatcher & Concurrency

Relevant source files

• internal/domain/workflow.go
• internal/domain/workflow_output.go
• internal/domain/workflow_run.go
• internal/repository/workflow.go
• internal/repository/workflow_output.go
• internal/repository/workflow_run.go
• internal/workflow/dispatcher/deps.go
• internal/workflow/dispatcher/dispatcher.go
• internal/workflow/engine/context.go
• internal/workflow/engine/engine.go
• internal/workflow/engine/executor_bizapply.go
• internal/workflow/engine/executor_bizdeploy.go
• internal/workflow/engine/executor_bizmonitor.go
• internal/workflow/engine/executor_biznotify.go
• internal/workflow/engine/executor_bizupload.go
• internal/workflow/engine/executor_condition.go
• internal/workflow/engine/executor_trycatch.go
• internal/workflow/engine/state.go
• internal/workflow/service.go
• ui/src/domain/workflow.ts
• ui/src/stores/workflow/index.ts

The Workflow Dispatcher is the central orchestration component responsible for managing the execution lifecycle of workflows. It handles task queuing, enforces global and per-workflow concurrency limits, manages execution states (pending, processing, canceled), and provides a bridge between the persistent repository layer and the volatile execution engine.

Core Architecture

The dispatcher operates as a singleton service defined by the WorkflowDispatcher interface [internal/workflow/dispatcher/dispatcher.go:33-40]. It is responsible for the transition of workflow runs from a database record to an active execution in the WorkflowEngine.

Concurrency Control

Certimate implements a two-tier concurrency model:

1. Global Concurrency Limit: Controlled by the environment variable CERTIMATE_WORKFLOW_MAX_WORKERS. It defaults to the number of available CPUs via runtime.GOMAXPROCS(0) [internal/workflow/dispatcher/dispatcher.go:24-27]. If the environment variable is set to 0 or less, it defaults to runtime.NumCPU() with a minimum of 1 [internal/workflow/dispatcher/dispatcher.go:28-30].
2. Per-Workflow Serialization: While the system can process multiple different workflows in parallel up to the global limit, the dispatcher logic ensures that individual runs are picked from the queue based on worker availability. The StartRun method in WorkflowService also prevents manual triggers if a workflow is already in a pending or processing state [internal/workflow/service.go:83-84].

Internal State Tracking

The dispatcher tracks tasks using the following structures within the workflowDispatcher struct [internal/workflow/dispatcher/dispatcher.go:48-61]:

• pendingRunQueue: A slice of RunId strings representing tasks waiting for an available worker [internal/workflow/dispatcher/dispatcher.go:53].
• processingTasks: A map keyed by RunId pointing to taskInfo, which contains the execution context, workflow ID, and a context.CancelFunc [internal/workflow/dispatcher/dispatcher.go:54].
• taskMtx: A sync.RWMutex used to protect access to the queue and task map during concurrent operations [internal/workflow/dispatcher/dispatcher.go:52].

Code Entity Space: Dispatcher Components

The following diagram illustrates how the logical dispatcher components map to specific Go structures and variables.

Title: Dispatcher Code Entity Map

Sources: [internal/workflow/dispatcher/dispatcher.go:24-31], [internal/workflow/dispatcher/dispatcher.go:48-61], [internal/workflow/dispatcher/dispatcher.go:244-245]

---

Lifecycle Management

The dispatcher manages three primary lifecycle phases: Bootup, Start, and Cancel.

1. Bootup & Recovery

When the application starts, the Bootup function is called [internal/workflow/dispatcher/dispatcher.go:84]. It performs a recovery of the database to ensure no tasks are stuck in a zombie state from a previous crash.

• It calls workflowRunRepo.ResetStatusIfHanging(ctx) [internal/workflow/dispatcher/dispatcher.go:92].
• The repository implementation executes SQL updates to move any pending or processing records to canceled status [internal/repository/workflow_run.go:160-190].
• This ensures that upon restart, the system has a clean slate and avoids double-processing of interrupted tasks.

2. Task Dispatching (Start)

When a workflow is triggered (manually or via cron), the Start method is invoked [internal/workflow/dispatcher/dispatcher.go:120]:

1. Acquires taskMtx lock [internal/workflow/dispatcher/dispatcher.go:121-122].
2. Checks if the runId is already in the processingTasks map or pendingRunQueue to prevent duplicate execution [internal/workflow/dispatcher/dispatcher.go:124-132].
3. Appends the runId to pendingRunQueue [internal/workflow/dispatcher/dispatcher.go:134].
4. Spawns a goroutine for tryNextAsync() to evaluate if a worker is available to pick up the task [internal/workflow/dispatcher/dispatcher.go:135].

3. Cancellation

The Cancel method allows for graceful termination of a running or pending task [internal/workflow/dispatcher/dispatcher.go:140]:

• If the task is in the pendingRunQueue, it is spliced out of the list [internal/workflow/dispatcher/dispatcher.go:176-181].
• If the task is in processingTasks, the dispatcher retrieves the taskInfo and calls the associated context.CancelFunc via task.cancel() [internal/workflow/dispatcher/dispatcher.go:170].
• The status in the repository is updated to canceled using workflowRunRepo.Save or SaveWithCascading [internal/workflow/dispatcher/dispatcher.go:158-167].

Sources: [internal/workflow/dispatcher/dispatcher.go:84-99], [internal/workflow/dispatcher/dispatcher.go:120-138], [internal/workflow/dispatcher/dispatcher.go:140-186], [internal/repository/workflow_run.go:160-190]

---

Execution Flow & Logging

The transition from the dispatcher to the engine occurs in tryExecuteAsync [internal/workflow/dispatcher/dispatcher.go:188]. This method is responsible for setting up the execution environment and handling the terminal state of a run.

Title: Workflow Execution Data Flow

Log Piping

The dispatcher pipes logs from the engine back to the database in real-time. It initializes the WorkflowEngine and attaches a logging hook [internal/workflow/dispatcher/dispatcher.go:244-253]:

• Engine Hooking: The dispatcher registers a callback using we.OnNodeLogging [internal/workflow/dispatcher/dispatcher.go:252].
• Persistence: For every log record produced by a node executor, the dispatcher creates a domain.WorkflowLog entity and persists it via workflowLogRepo.Save [internal/workflow/dispatcher/dispatcher.go:254-257].

Panic Recovery

The dispatcher includes a robust recovery mechanism in tryExecuteAsync. If the engine panics, the dispatcher:

1. Captures the panic using recover() in a deferred block [internal/workflow/dispatcher/dispatcher.go:194-208].
2. Logs the stack trace using debug.Stack() [internal/workflow/dispatcher/dispatcher.go:197].
3. Updates the WorkflowRun status to failed and saves the panic message to the Error field [internal/workflow/dispatcher/dispatcher.go:200-205].

Sources: [internal/workflow/dispatcher/dispatcher.go:188-217], [internal/workflow/dispatcher/dispatcher.go:243-258]

---

Concurrency Statistics

The dispatcher provides a GetStatistics method to monitor the current state of the engine. This is exposed via WorkflowService.GetStatistics [internal/workflow/service.go:68-75].

Metric	Description	Source
Concurrency	The maximum allowed workers (CERTIMATE_WORKFLOW_MAX_WORKERS)	wd.concurrency [internal/workflow/dispatcher/dispatcher.go:50]
PendingRunIds	List of Run IDs waiting in the pendingRunQueue	wd.pendingRunQueue [internal/workflow/dispatcher/dispatcher.go:53]
ProcessingRunIds	List of Run IDs currently being executed in processingTasks	wd.processingTasks [internal/workflow/dispatcher/dispatcher.go:54]

Sources: [internal/workflow/dispatcher/dispatcher.go:42-46], [internal/workflow/dispatcher/dispatcher.go:65-82], [internal/workflow/service.go:68-75]