Skip to main content

Workflow & Run Management

Flyte manages the lifecycle of workflow runs through a combination of gRPC services, persistent storage in PostgreSQL, and background reconciliation. The system tracks every execution from the initial trigger to final termination, maintaining a detailed history of state transitions and cluster events.

Workflow Run Lifecycle​

When you trigger a workflow, Flyte initiates a run that progresses through several phases (e.g., QUEUED, RUNNING, SUCCEEDED, FAILED). The primary entry point for these operations is the RunService located in runs/service/run_service.go.

Creating a Run​

To start a new execution, you call CreateRun. Internally, RunService performs the following steps:

  1. ID Generation: Generates a unique RunIdentifier using the project, domain, and a timestamp-based name.
  2. Persistence: Calls persistRunModel to save the run metadata and the initial ActionSpec to the database via the ActionRepo.
  3. Execution Enqueue: Submits the run to the execution engine by calling s.actionsClient.Enqueue.
// From runs/service/run_service.go
func (s *RunService) CreateRun(
	ctx context.Context,
	req *connect.Request[workflow.CreateRunRequest],
) (*connect.Response[workflow.CreateRunResponse], error) {
    // ... validation and ID generation ...
    runId = &common.RunIdentifier{
        Project: id.ProjectId.Name,
        Domain:  id.ProjectId.Domain,
        Name:    generateRunName(time.Now().UnixNano()),
    }
    // ... storage and DB persistence ...
    run, err := s.persistRunModel(ctx, runId, taskID, taskSpec, inputPrefix, runOutputBase, runSpec, request.GetSource(), triggerName, triggerTaskName, triggerRevision, triggerType)
    
    // Enqueue to execution engine
    _, err = s.actionsClient.Enqueue(ctx, connect.NewRequest(&actions.EnqueueRequest{
        Action: &actions.Action{
            ActionId:      actionID,
            InputUri:      inputPrefix,
            RunOutputBase: runOutputBase,
            Spec: &actions.Action_Task{
                Task: &workflow.TaskAction{
                    Id:       taskID,
                    Spec:     taskSpec,
                    CacheKey: wrapperspb.String(cacheKey),
                },
            },
        },
        RunSpec: runSpec,
    }))
    // ...
}

State Persistence and the Root Action​

Flyte represents the top-level workflow run as a special action named a0. All subsequent tasks or sub-workflows within that run are stored as child actions in the actions table. The ActionRepo (implemented in runs/repository/impl/action.go) handles the CRUD operations for these entities.

When an action's phase changes, UpdateActionPhase updates the database and triggers a notification:

  • If the root action a0 is updated, it notifies subscribers of a Run update.
  • For any action, it notifies subscribers of an Action update.

Reliable Termination​

If you need to stop a running workflow, you call AbortRun. However, simply marking a run as ABORTED in the database does not immediately stop the underlying Kubernetes pods or external resources.

The Abort Reconciler​

Flyte uses the AbortReconciler (found in runs/service/abort_reconciler.go) to ensure that abort requests are reliably propagated to the execution engine.

  1. AbortRun marks the root action a0 as ABORTED in the database.
  2. The AbortReconciler background service polls for actions in a pending abort state.
  3. It uses a deduplicated work queue and exponential backoff to retry termination calls (actionsClient.Abort) until the execution engine confirms the action has stopped.

You can configure the reconciler's behavior via AbortReconcilerConfig:

  • Workers: Number of concurrent termination goroutines.
  • MaxAttempts: Maximum retries per action before giving up.
  • QueueSize: Buffer size for the internal work channel.

Real-time State Tracking​

Flyte provides streaming APIs (WatchRuns and WatchActions) to allow frontends and CLI tools to track execution progress in real-time without polling.

LISTEN/NOTIFY Mechanism​

The ActionRepo utilizes PostgreSQL's LISTEN and NOTIFY commands to broadcast updates. When UpdateActionPhase is called, it executes a NOTIFY on a specific channel. The Watch methods in RunService subscribe to these channels to receive immediate updates.

Tree-based State Management​

For complex workflows with many nested tasks, Flyte uses an internal runStateManager (runs/service/run_state_manager.go) to maintain a tree-based view of the run. This manager:

  • Aggregates Phase Counts: Tracks how many children are in each phase (e.g., 5 succeeded, 2 running) to provide summary statistics for parent nodes.
  • Handles Filtering: Efficiently determines which nodes should be visible based on user-provided filters (e.g., "show only failed tasks").
  • Maintains Visibility: Ensures that if a child matches a filter, all its ancestors remain visible in the tree so the user can see the full path.
// From runs/service/run_state_manager.go
func (rsm *runStateManager) modifyPhaseCounters(current *node, toPhase, fromPhase common.ActionPhase, changed map[string]struct{}) {
    for current != nil {
        if toPhase != common.ActionPhase_ACTION_PHASE_UNSPECIFIED {
            current.ChildPhaseCounts[toPhase]++
        }
        if fromPhase != common.ActionPhase_ACTION_PHASE_UNSPECIFIED {
            count := current.ChildPhaseCounts[fromPhase]
            current.ChildPhaseCounts[fromPhase] = max(0, count-1)
        }
        changed[current.Action.Name] = struct{}{}
        current = current.Parent
    }
}

Event History and Cluster Events​

Granular execution details, such as Kubernetes pod events or container logs, are recorded as ActionEvents.

Internal Run Service​

The InternalRunService (in runs/service/internal_run_service.go) provides the RecordActionEvent endpoint used by the execution engine to report progress. These events are stored in the action_events table and linked to specific action attempts.

Watching Cluster Events​

You can stream these low-level events using WatchClusterEvents. This method fetches existing events from the database and then waits for new ones via the same ActionRepo notification mechanism used for phase updates. This allows you to see pod scheduling, image pulling, and container start/stop events as they happen.

// From runs/service/run_service.go
func (s *RunService) WatchClusterEvents(
	ctx context.Context,
	req *connect.Request[workflow.WatchClusterEventsRequest],
	stream *connect.ServerStream[workflow.WatchClusterEventsResponse],
) error {
    // ...
    // Start watching first to reduce the chance of missing action updates
    updatesCh := make(chan *models.Action, 50)
    go s.repo.ActionRepo().WatchActionUpdates(ctx, actionID, updatesCh, errsCh)

    // Drain all available cluster events for current action
    info, err := s.getClusterEventsInfo(ctx, actionID, attempt, lastUpdatedAt, offset, maxEvents)
    // ...
}

Eventual Consistency Gotcha​

Note that event recording is eventually consistent. If you call GetRunDetails immediately after an event is triggered in the cluster, it may take a few moments for the InternalRunService to process the event and for it to appear in the API response.