fix: remove pipeline run promotion blocking logic

Removed the mechanism that permanently blocked pipeline runs from being
promoted after a set number of failed attempts. This ensures that
transient failures do not cause pipeline runs to get stuck indefinitely,
allowing the system to recover automatically once the underlying issue
is resolved.

Signed-off-by: Chmouel Boudjnah <chmouel@redhat.com>

Author: chmouel

pkg/queue/common.go

@@ -1,7 +1,13 @@
 package queue
 
 import (
+	"slices"
+	"strings"
 	"time"
+
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/keys"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/kubeinteraction"
+	tektonv1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1"
 )
 
 type Semaphore interface {
@@ -18,3 +24,42 @@ type Semaphore interface {
 	getCurrentRunning() []string
 	getCurrentPending() []string
 }
+
+func IsRecoverableQueuedPipelineRun(pr *tektonv1.PipelineRun) bool {
+	if pr == nil {
+		return false
+	}
+	if pr.GetAnnotations()[keys.State] != kubeinteraction.StateQueued {
+		return false
+	}
+	if pr.Spec.Status != tektonv1.PipelineRunSpecStatusPending {
+		return false
+	}
+	if pr.IsDone() || pr.IsCancelled() {
+		return false
+	}
+	_, ok := executionOrderIndex(pr)
+	return ok
+}
+
+func executionOrderList(pr *tektonv1.PipelineRun) []string {
+	order := pr.GetAnnotations()[keys.ExecutionOrder]
+	if order == "" {
+		return nil
+	}
+	return strings.Split(order, ",")
+}
+
+func executionOrderIndex(pr *tektonv1.PipelineRun) (int, bool) {
+	order := executionOrderList(pr)
+	if len(order) == 0 {
+		return 0, false
+	}
+
+	key := PrKey(pr)
+	index := slices.Index(order, key)
+	if index < 0 {
+		return 0, false
+	}
+	return index, true
+}

pkg/queue/lease_manager.go

@@ -6,7 +6,6 @@ import (
 	"encoding/hex"
 	"fmt"
 	"os"
-	"slices"
 	"sort"
 	"strings"
 	"time"
@@ -264,10 +263,7 @@ func (m *LeaseManager) getRepoQueueState(ctx context.Context, repo *v1alpha1.Rep
 				state.running = append(state.running, pr)
 			}
 		case kubeinteraction.StateQueued:
-			if pr.Spec.Status != tektonv1.PipelineRunSpecStatusPending || pr.IsDone() || pr.IsCancelled() {
-				continue
-			}
-			if pr.GetAnnotations()[keys.QueuePromotionBlocked] == "true" {
+			if !IsRecoverableQueuedPipelineRun(&pr) {
 				continue
 			}
 			position, ok := executionOrderIndex(&pr)
@@ -341,28 +337,6 @@ func comparePipelineRuns(left, right *tektonv1.PipelineRun) bool {
 	return left.CreationTimestamp.Before(&right.CreationTimestamp)
 }
 
-func executionOrderList(pr *tektonv1.PipelineRun) []string {
-	order := pr.GetAnnotations()[keys.ExecutionOrder]
-	if order == "" {
-		return nil
-	}
-	return strings.Split(order, ",")
-}
-
-func executionOrderIndex(pr *tektonv1.PipelineRun) (int, bool) {
-	order := executionOrderList(pr)
-	if len(order) == 0 {
-		return 0, false
-	}
-
-	key := PrKey(pr)
-	index := slices.Index(order, key)
-	if index < 0 {
-		return 0, false
-	}
-	return index, true
-}
-
 func (m *LeaseManager) hasActiveClaim(pr *tektonv1.PipelineRun, now time.Time) bool {
 	annotations := pr.GetAnnotations()
 	claimedBy := annotations[keys.QueueClaimedBy]

pkg/queue/lease_manager_test.go

@@ -140,7 +140,7 @@ func TestLeaseManagerIgnoresMalformedExecutionOrder(t *testing.T) {
 	assert.DeepEqual(t, claimed, []string{"test-ns/bbb"})
 }
 
-func TestLeaseManagerIgnoresBlockedQueuedRuns(t *testing.T) {
+func TestLeaseManagerStillConsidersPreviouslyBlockedQueuedRuns(t *testing.T) {
 	ctx, _ := rtesting.SetupFakeContext(t)
 	observer, _ := zapobserver.New(zap.InfoLevel)
 	logger := zap.New(observer).Sugar()
@@ -160,7 +160,7 @@ func TestLeaseManagerIgnoresBlockedQueuedRuns(t *testing.T) {
 
 	claimed, err := manager.AddListToRunningQueue(ctx, repo, []string{"test-ns/aaa", "test-ns/bbb"})
 	assert.NilError(t, err)
-	assert.DeepEqual(t, claimed, []string{"test-ns/bbb"})
+	assert.DeepEqual(t, claimed, []string{"test-ns/aaa"})
 }
 
 func TestLeaseManagerReleaseKeepsLeaseAndClearsHolder(t *testing.T) {

pkg/reconciler/controller.go

@@ -181,16 +181,7 @@ func isEligibleLeaseQueueRecoveryCandidate(pipelineRun *tektonv1.PipelineRun) bo
 	if pipelineRun.GetAnnotations()[keys.Repository] == "" {
 		return false
 	}
-	if pipelineRun.GetAnnotations()[keys.State] != kubeinteraction.StateQueued {
-		return false
-	}
-	if pipelineRun.Spec.Status != tektonv1.PipelineRunSpecStatusPending {
-		return false
-	}
-	if pipelineRun.IsDone() || pipelineRun.IsCancelled() {
-		return false
-	}
-	return pipelineRun.GetAnnotations()[keys.QueuePromotionBlocked] != "true"
+	return queuepkg.IsRecoverableQueuedPipelineRun(pipelineRun)
 }
 
 func shouldPreferLeaseQueueRecoveryCandidate(left, right *tektonv1.PipelineRun) bool {

pkg/reconciler/controller_test.go

@@ -100,19 +100,24 @@ func TestSelectLeaseQueueRecoveryKeys(t *testing.T) {
 			pipelineRuns: []*pipelinev1.PipelineRun{
 				newLeaseRecoveryPR("later", "test-ns", "repo-a", now.Add(2*time.Second), map[string]string{}, false),
 				newLeaseRecoveryPR("earlier", "test-ns", "repo-a", now.Add(time.Second), map[string]string{}, false),
-				newLeaseRecoveryPR("blocked", "test-ns", "repo-b", now, map[string]string{
-					keys.QueuePromotionBlocked: "true",
+				newLeaseRecoveryPR("missing-order", "test-ns", "repo-b", now, map[string]string{
+					keys.ExecutionOrder: "",
+				}, false),
+				newLeaseRecoveryPR("malformed-order", "test-ns", "repo-c", now.Add(250*time.Millisecond), map[string]string{
+					keys.ExecutionOrder: "test-ns/some-other-pr",
 				}, false),
 				newLeaseRecoveryPR("valid", "test-ns", "repo-b", now.Add(3*time.Second), map[string]string{}, false),
+				newLeaseRecoveryPR("valid-after-malformed", "test-ns", "repo-c", now.Add(4*time.Second), map[string]string{}, false),
 				newLeaseRecoveryPR("other-namespace", "other-ns", "repo-a", now.Add(4*time.Second), map[string]string{}, false),
 				newLeaseRecoveryPR("missing-repo", "test-ns", "", now.Add(5*time.Second), map[string]string{}, false),
-				newLeaseRecoveryStartedPR("started", "test-ns", "repo-c", now.Add(6*time.Second)),
+				newLeaseRecoveryStartedPR("started", "test-ns", "repo-d", now.Add(6*time.Second)),
 				newLeaseRecoveryPR("done", "test-ns", "repo-d", now.Add(7*time.Second), map[string]string{}, true),
 			},
 			want: []string{
 				"test-ns/earlier",
 				"test-ns/valid",
 				"other-ns/other-namespace",
+				"test-ns/valid-after-malformed",
 			},
 		},
 	}
@@ -163,7 +168,8 @@ func newLeaseRecoveryPR(
 	done bool,
 ) *pipelinev1.PipelineRun {
 	annotations := map[string]string{
-		keys.State: kubeinteraction.StateQueued,
+		keys.State:          kubeinteraction.StateQueued,
+		keys.ExecutionOrder: namespace + "/" + name,
 	}
 	if repo != "" {
 		annotations[keys.Repository] = repo

pkg/reconciler/queue_pipelineruns.go

@@ -17,8 +17,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-const maxQueuePromotionRetries = 5
-
 func (r *Reconciler) queuePipelineRun(ctx context.Context, logger *zap.SugaredLogger, pr *tektonv1.PipelineRun) error {
 	order, exist := pr.GetAnnotations()[keys.ExecutionOrder]
 	if !exist {
@@ -99,17 +97,14 @@ func (r *Reconciler) queuePipelineRun(ctx context.Context, logger *zap.SugaredLo
 						continue
 					}
 					logger.Errorf("failed to update pipelineRun to in_progress: %w", err)
-					blocked, retryErr := r.recordQueuePromotionFailure(ctx, logger, pr, err)
+					retryErr := r.recordQueuePromotionFailure(ctx, logger, pr, err)
 					if retryErr != nil {
-						logger.Warnf("failed to record queue promotion failure for %s: %v", pr.GetName(), retryErr)
-					}
-					if blocked {
-						logger.Errorf("blocking further queue promotion attempts for pipelineRun %s after %d failures", pr.GetName(), maxQueuePromotionRetries)
+						return fmt.Errorf("failed to record queue promotion failure for %s after promotion error: %w", pr.GetName(), retryErr)
 					}
 					_ = r.qm.RemoveFromQueue(ctx, repo, prKeys)
-				} else {
-					processed = true
+					return fmt.Errorf("failed to update pipelineRun to in_progress: %w", err)
 				}
+				processed = true
 			}
 		}
 		if processed {
@@ -131,7 +126,7 @@ func (r *Reconciler) pipelineRunReachedStartedState(ctx context.Context, pr *tek
 	return latest.GetAnnotations()[keys.State] == kubeinteraction.StateStarted, nil
 }
 
-func (r *Reconciler) recordQueuePromotionFailure(ctx context.Context, logger *zap.SugaredLogger, pr *tektonv1.PipelineRun, cause error) (bool, error) {
+func (r *Reconciler) recordQueuePromotionFailure(ctx context.Context, logger *zap.SugaredLogger, pr *tektonv1.PipelineRun, cause error) error {
 	retries := 0
 	if current := pr.GetAnnotations()[keys.QueuePromotionRetries]; current != "" {
 		parsed, err := strconv.Atoi(current)
@@ -145,18 +140,14 @@ func (r *Reconciler) recordQueuePromotionFailure(ctx context.Context, logger *za
 		keys.QueuePromotionRetries: strconv.Itoa(retries),
 		keys.QueuePromotionLastErr: cause.Error(),
 	}
-	blocked := retries >= maxQueuePromotionRetries
-	if blocked {
-		annotations[keys.QueuePromotionBlocked] = "true"
-	}
 
 	_, err := action.PatchPipelineRun(ctx, logger, "queue promotion failure", r.run.Clients.Tekton, pr, map[string]any{
 		"metadata": map[string]any{
 			"annotations": annotations,
 		},
 	})
 	if err != nil {
-		return false, err
+		return err
 	}
-	return blocked, nil
+	return nil
 }

pkg/reconciler/queue_pipelineruns_test.go

@@ -3,10 +3,12 @@ package reconciler
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/keys"
 	pacv1alpha1 "github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/v1alpha1"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/kubeinteraction"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/params"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/clients"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/info"
@@ -17,6 +19,8 @@ import (
 	zapobserver "go.uber.org/zap/zaptest/observer"
 	"gotest.tools/v3/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	k8stesting "k8s.io/client-go/testing"
 	rtesting "knative.dev/pkg/reconciler/testing"
 )
 
@@ -215,26 +219,21 @@ func TestRecordQueuePromotionFailure(t *testing.T) {
 	fakelogger := zap.New(observer).Sugar()
 
 	tests := []struct {
-		name              string
-		annotations       map[string]string
-		wantRetries       string
-		wantBlocked       bool
-		wantBlockedMarker string
+		name        string
+		annotations map[string]string
+		wantRetries string
 	}{
 		{
 			name:        "first failure records retry metadata",
 			annotations: map[string]string{},
 			wantRetries: "1",
-			wantBlocked: false,
 		},
 		{
-			name: "reaching retry limit blocks future promotion",
+			name: "later failures keep incrementing retries without blocking promotion",
 			annotations: map[string]string{
 				keys.QueuePromotionRetries: "4",
 			},
-			wantRetries:       "5",
-			wantBlocked:       true,
-			wantBlockedMarker: "true",
+			wantRetries: "5",
 		},
 	}
 
@@ -261,21 +260,92 @@ func TestRecordQueuePromotionFailure(t *testing.T) {
 				},
 			}
 
-			blocked, err := r.recordQueuePromotionFailure(ctx, fakelogger, pipelineRun, errors.New("cannot patch"))
+			err := r.recordQueuePromotionFailure(ctx, fakelogger, pipelineRun, errors.New("cannot patch"))
 			assert.NilError(t, err)
-			assert.Equal(t, blocked, tt.wantBlocked)
 
 			updatedPR, err := stdata.Pipeline.TektonV1().PipelineRuns("test").Get(ctx, "test-pr", metav1.GetOptions{})
 			assert.NilError(t, err)
 			assert.Equal(t, updatedPR.GetAnnotations()[keys.QueuePromotionRetries], tt.wantRetries)
 			assert.Equal(t, updatedPR.GetAnnotations()[keys.QueuePromotionLastErr], "cannot patch")
-
-			if tt.wantBlockedMarker == "" {
-				_, exists := updatedPR.GetAnnotations()[keys.QueuePromotionBlocked]
-				assert.Assert(t, !exists, "QueuePromotionBlocked should not be set before the retry limit is reached")
-			} else {
-				assert.Equal(t, updatedPR.GetAnnotations()[keys.QueuePromotionBlocked], tt.wantBlockedMarker)
-			}
+			_, exists := updatedPR.GetAnnotations()[keys.QueuePromotionBlocked]
+			assert.Assert(t, !exists, "QueuePromotionBlocked should not be set when promotion fails")
 		})
 	}
 }
+
+func TestQueuePipelineRunStopsAfterSinglePromotionFailure(t *testing.T) {
+	observer, _ := zapobserver.New(zap.InfoLevel)
+	fakelogger := zap.New(observer).Sugar()
+	ctx, _ := rtesting.SetupFakeContext(t)
+
+	pipelineRun := &tektonv1.PipelineRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "queued",
+			Namespace: "test",
+			Annotations: map[string]string{
+				keys.ExecutionOrder: "test/queued",
+				keys.Repository:     "test",
+				keys.State:          kubeinteraction.StateQueued,
+			},
+		},
+		Spec: tektonv1.PipelineRunSpec{
+			Status: tektonv1.PipelineRunSpecStatusPending,
+		},
+	}
+	testRepo := &pacv1alpha1.Repository{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test",
+			Namespace: "test",
+		},
+		Spec: pacv1alpha1.RepositorySpec{
+			URL: randomURL,
+		},
+	}
+
+	testData := testclient.Data{
+		Repositories: []*pacv1alpha1.Repository{testRepo},
+		PipelineRuns: []*tektonv1.PipelineRun{pipelineRun},
+	}
+	stdata, informers := testclient.SeedTestData(t, ctx, testData)
+	patchCalls := 0
+	stdata.Pipeline.PrependReactor("patch", "pipelineruns", func(_ k8stesting.Action) (bool, runtime.Object, error) {
+		patchCalls++
+		if patchCalls == 1 {
+			return true, nil, errors.New("boom")
+		}
+		return false, nil, nil
+	})
+
+	r := &Reconciler{
+		qm: testconcurrency.TestQMI{
+			RunningQueue: []string{"test/queued"},
+		},
+		repoLister: informers.Repository.Lister(),
+		run: &params.Run{
+			Info: info.Info{
+				Kube: &info.KubeOpts{
+					Namespace: "global",
+				},
+				Controller: &info.ControllerInfo{},
+			},
+			Clients: clients.Clients{
+				PipelineAsCode: stdata.PipelineAsCode,
+				Tekton:         stdata.Pipeline,
+				Kube:           stdata.Kube,
+				Log:            fakelogger,
+			},
+		},
+	}
+
+	err := r.queuePipelineRun(ctx, fakelogger, pipelineRun)
+	assert.ErrorContains(t, err, "failed to update pipelineRun to in_progress")
+
+	updatedPR, getErr := stdata.Pipeline.TektonV1().PipelineRuns("test").Get(ctx, "queued", metav1.GetOptions{})
+	assert.NilError(t, getErr)
+	assert.Equal(t, updatedPR.GetAnnotations()[keys.QueuePromotionRetries], "1")
+	assert.Assert(t, updatedPR.GetAnnotations()[keys.QueuePromotionLastErr] != "")
+	assert.Assert(t, strings.Contains(updatedPR.GetAnnotations()[keys.QueuePromotionLastErr], "cannot update state"))
+	assert.Assert(t, strings.Contains(updatedPR.GetAnnotations()[keys.QueuePromotionLastErr], "boom"))
+	_, exists := updatedPR.GetAnnotations()[keys.QueuePromotionBlocked]
+	assert.Assert(t, !exists, "QueuePromotionBlocked should not be set when queuePipelineRun returns after a failed promotion")
+}