Aborting and Reconciling Executions

When a user requests to cancel a run or a specific action in Flyte, the system must ensure that the underlying compute resources (such as Kubernetes pods) are actually terminated. Flyte implements this using a two-step process: first, marking the intent in the database, and second, using a background reconciler to drive the termination to completion.

Triggering an Abort Request

The RunService handles incoming abort requests by updating the database and then handing off the task to the AbortReconciler. For a full run cancellation, Flyte specifically targets the root action (a0), relying on Kubernetes owner references to cascade the deletion to all child actions.

// From runs/service/run_service.go

func (s *RunService) AbortRun(
	ctx context.Context,
	req *connect.Request[workflow.AbortRunRequest],
) (*connect.Response[workflow.AbortRunResponse], error) {
	// ... validation ...
	reason := "User requested abort"
	if req.Msg.Reason != nil {
		reason = *req.Msg.Reason
	}

	// Mark only the root action ABORTED in DB, then push it to the reconciler.
	if err := s.repo.ActionRepo().AbortRun(ctx, req.Msg.RunId, reason, nil); err != nil {
		return nil, connect.NewError(connect.CodeInternal, err)
	}

	if s.abortReconciler != nil {
		// "a0" is the root action for every run
		s.abortReconciler.Push(ctx, &common.ActionIdentifier{Run: req.Msg.RunId, Name: "a0"}, reason)
	}

	return connect.NewResponse(&workflow.AbortRunResponse{}), nil
}

The Reconciliation Loop

The AbortReconciler runs as a background worker. It consumes tasks from an internal queue and calls the ActionsServiceClient.Abort method. If the call fails, it schedules a retry with exponential backoff.

The core logic resides in processTask, which manages the lifecycle of an abort attempt:

// From runs/service/abort_reconciler.go

func (r *AbortReconciler) processTask(ctx context.Context, task abortTask) {
	// Increment attempt count in DB
	attemptCount, err := r.repo.ActionRepo().MarkAbortAttempt(ctx, task.actionID)
	if err != nil {
		logger.Errorf(ctx, "AbortReconciler: failed to mark attempt for %s: %v", task.key, err)
		r.queue.remove(task.key)
		return
	}

	// Call the Actions service to terminate the pod/resource
	_, abortErr := r.actionsClient.Abort(ctx, connect.NewRequest(&actions.AbortRequest{
		ActionId: task.actionID,
		Reason:   &task.reason,
	}))

	// Success or "Already Gone" cases
	if abortErr == nil || isAlreadyTerminated(abortErr) {
		if clearErr := r.repo.ActionRepo().ClearAbortRequest(ctx, task.actionID); clearErr != nil {
			logger.Errorf(ctx, "AbortReconciler: failed to clear abort request for %s: %v", task.key, clearErr)
		}
		r.queue.remove(task.key)
		return
	}

	// Handle retries or give up
	if attemptCount >= r.cfg.MaxAttempts {
		logger.Errorf(ctx, "AbortReconciler: giving up on %s after %d attempts", task.key, attemptCount)
		r.repo.ActionRepo().ClearAbortRequest(ctx, task.actionID)
		r.queue.remove(task.key)
		return
	}

	r.scheduleRetry(ctx, task, attemptCount)
}

Reliability and Crash Recovery

Flyte ensures that abort requests are not lost if the service restarts or if multiple requests are made for the same action.

Deduplication

The dedupeQueue prevents redundant work. If an abort request for a specific action is already being processed or is waiting in the retry backoff, subsequent Push calls for that same action are ignored.

// From runs/service/abort_reconciler.go

type dedupeQueue struct {
	mu   sync.Mutex
	keys map[string]struct{}
	ch   chan abortTask
}

func (q *dedupeQueue) push(task abortTask) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	if _, ok := q.keys[task.key]; ok {
		return false // Already in queue or being processed
	}
	q.keys[task.key] = struct{}{}
	q.ch <- task
	return true
}

Startup Recovery

When the AbortReconciler starts, it performs a startupScan. It queries the database for any actions that have an abort_requested_at timestamp but haven't been cleared yet, ensuring that pending cancellations are resumed after a service restart.

func (r *AbortReconciler) startupScan(ctx context.Context) error {
	pending, err := r.repo.ActionRepo().ListPendingAborts(ctx)
	if err != nil {
		return err
	}
	for _, p := range pending {
		r.Push(ctx, p.ActionID, p.Reason)
	}
	return nil
}

Configuring the Reconciler

You can tune the reconciler's behavior using the AbortReconcilerConfig. These settings control the concurrency and the aggressiveness of retries.

Parameter	Description	Default
Workers	Number of concurrent goroutines processing aborts.	5
MaxAttempts	Maximum retries before giving up on an action.	10
QueueSize	Buffer size for the internal task channel.	1000
InitialDelay	Starting backoff duration for retries.	1s
MaxDelay	Maximum backoff duration.	5m

Troubleshooting Termination Failures

"Not Found" Errors

The reconciler treats "Not Found" errors from the Actions service as a success. This is because a "Not Found" error typically means the resource (e.g., the Kubernetes pod) has already been deleted. Flyte checks both the gRPC status code and the error message to identify these cases:

func isAlreadyTerminated(err error) bool {
	connectErr, ok := err.(*connect.Error)
	if !ok {
		return false
	}
	// Check for standard NotFound or Internal errors containing "not found"
	if connectErr.Code() == connect.CodeNotFound || 
	   (connectErr.Code() == connect.CodeInternal && strings.Contains(connectErr.Message(), "not found")) {
		return true
	}
	return false
}

Manual Intervention

If an action reaches MaxAttempts without success, the reconciler logs an error and clears the request from the database to prevent the queue from filling up indefinitely. In these cases, the underlying resource may still be running, and manual intervention in the cluster (e.g., kubectl delete pod) may be required.