feat: Implement background recovery for queued PipelineRuns

Introduced a background recovery mechanism for the lease manager to
periodically re-enqueue PipelineRuns that remained in a queued state.
This ensured that tasks did not get stuck if a lease was lost or if
the controller restarted. The loop identified eligible pending runs
for each repository and added them back to the work queue.

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

Author: chmouel

pkg/params/settings/config.go

@@ -84,7 +84,7 @@ type Settings struct {
 
 	RememberOKToTest   bool   `json:"remember-ok-to-test"`
 	RequireOkToTestSHA bool   `json:"require-ok-to-test-sha"`
-	ConcurrencyBackend string `default:"memory" json:"concurrency-backend"`
+	ConcurrencyBackend string `default:"memory"              json:"concurrency-backend"`
 }
 
 func (s *Settings) DeepCopy(out *Settings) {

pkg/queue/lease_manager.go

@@ -2,7 +2,7 @@ package queue
 
 import (
 	"context"
-	"crypto/sha1" //nolint:gosec
+	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
 	"os"
@@ -86,6 +86,10 @@ func (m *LeaseManager) InitQueues(context.Context, tektonclient.Interface, versi
 	return nil
 }
 
+func (m *LeaseManager) RecoveryInterval() time.Duration {
+	return m.claimTTL
+}
+
 func (m *LeaseManager) RemoveRepository(repo *v1alpha1.Repository) {
 	releaseCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -568,6 +572,6 @@ func (m *LeaseManager) releaseLease(leaseName string) {
 }
 
 func repoLeaseName(repoKey string) string {
-	sum := sha1.Sum([]byte(repoKey))
+	sum := sha256.Sum256([]byte(repoKey))
 	return fmt.Sprintf("pac-concurrency-%s", hex.EncodeToString(sum[:8]))
 }

pkg/queue/queue_manager.go

@@ -37,6 +37,10 @@ func NewManager(logger *zap.SugaredLogger) *Manager {
 	}
 }
 
+func (*Manager) RecoveryInterval() time.Duration {
+	return 0
+}
+
 // getSemaphore returns existing semaphore created for repository or create
 // a new one with limit provided in repository
 // Semaphore: nothing but a waiting and a running queue for a repository

pkg/queue/queue_manager_interface.go

@@ -3,6 +3,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/v1alpha1"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/generated/clientset/versioned"
@@ -12,6 +13,7 @@ import (
 
 type ManagerInterface interface {
 	InitQueues(ctx context.Context, tekton tektonVersionedClient.Interface, pac versioned.Interface) error
+	RecoveryInterval() time.Duration
 	RemoveRepository(repo *v1alpha1.Repository)
 	QueuedPipelineRuns(repo *v1alpha1.Repository) []string
 	RunningPipelineRuns(repo *v1alpha1.Repository) []string

pkg/queue/queue_manager_test.go

@@ -46,6 +46,17 @@ func TestSomeoneElseSetPendingWithNoConcurrencyLimit(t *testing.T) {
 	assert.Equal(t, len(started), 1)
 }
 
+func TestManagerRecoveryInterval(t *testing.T) {
+	observer, _ := zapobserver.New(zap.InfoLevel)
+	logger := zap.New(observer).Sugar()
+
+	memoryManager := NewManager(logger)
+	assert.Equal(t, memoryManager.RecoveryInterval(), time.Duration(0))
+
+	leaseManager := NewLeaseManager(logger, nil, nil, "pac")
+	assert.Equal(t, leaseManager.RecoveryInterval(), defaultLeaseClaimTTL)
+}
+
 func TestAddToPendingQueueDirectly(t *testing.T) {
 	observer, _ := zapobserver.New(zap.InfoLevel)
 	logger := zap.New(observer).Sugar()

pkg/reconciler/controller.go

@@ -3,6 +3,8 @@ package reconciler
 import (
 	"context"
 	"path"
+	"sort"
+	"time"
 
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/keys"
@@ -16,6 +18,9 @@ import (
 	tektonv1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1"
 	tektonPipelineRunInformerv1 "github.com/tektoncd/pipeline/pkg/client/injection/informers/pipeline/v1/pipelinerun"
 	tektonPipelineRunReconcilerv1 "github.com/tektoncd/pipeline/pkg/client/injection/reconciler/pipeline/v1/pipelinerun"
+	tektonv1lister "github.com/tektoncd/pipeline/pkg/client/listers/pipeline/v1"
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -24,6 +29,8 @@ import (
 	"knative.dev/pkg/system"
 )
 
+const leaseQueueRecoveryBuffer = time.Second
+
 func NewController() func(context.Context, configmap.Watcher) *controller.Impl {
 	return func(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 		ctx = info.StoreNS(ctx, system.Namespace())
@@ -71,13 +78,128 @@ func NewController() func(context.Context, configmap.Watcher) *controller.Impl {
 			logging.FromContext(ctx).Panicf("Couldn't register PipelineRun informer event handler: %w", err)
 		}
 
+		if recoveryInterval := r.qm.RecoveryInterval(); recoveryInterval > 0 {
+			startLeaseQueueRecoveryLoop(ctx, log, impl, r.pipelineRunLister, recoveryInterval+leaseQueueRecoveryBuffer)
+		}
+
 		// Start pac config syncer after the initial settings have been loaded.
 		go params.StartConfigSync(ctx, run)
 
 		return impl
 	}
 }
 
+func startLeaseQueueRecoveryLoop(
+	ctx context.Context,
+	logger *zap.SugaredLogger,
+	impl *controller.Impl,
+	lister tektonv1lister.PipelineRunLister,
+	interval time.Duration,
+) {
+	if interval <= 0 {
+		return
+	}
+
+	logger.Infof("starting lease queue recovery loop with interval %s", interval)
+	runLeaseQueueRecovery(logger, impl, lister)
+
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				runLeaseQueueRecovery(logger, impl, lister)
+			}
+		}
+	}()
+}
+
+func runLeaseQueueRecovery(logger *zap.SugaredLogger, impl *controller.Impl, lister tektonv1lister.PipelineRunLister) {
+	recoveryKeys, err := leaseQueueRecoveryKeys(lister)
+	if err != nil {
+		logger.Warnf("failed to list queued PipelineRuns for lease recovery: %v", err)
+		return
+	}
+
+	for _, key := range recoveryKeys {
+		impl.EnqueueKey(key)
+	}
+}
+
+func leaseQueueRecoveryKeys(lister tektonv1lister.PipelineRunLister) ([]types.NamespacedName, error) {
+	pipelineRuns, err := lister.List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+	return selectLeaseQueueRecoveryKeys(pipelineRuns), nil
+}
+
+func selectLeaseQueueRecoveryKeys(pipelineRuns []*tektonv1.PipelineRun) []types.NamespacedName {
+	candidatesByRepo := map[string]*tektonv1.PipelineRun{}
+
+	for _, pipelineRun := range pipelineRuns {
+		if !isEligibleLeaseQueueRecoveryCandidate(pipelineRun) {
+			continue
+		}
+
+		repoKey := types.NamespacedName{
+			Namespace: pipelineRun.GetNamespace(),
+			Name:      pipelineRun.GetAnnotations()[keys.Repository],
+		}.String()
+		if existing, ok := candidatesByRepo[repoKey]; !ok || shouldPreferLeaseQueueRecoveryCandidate(pipelineRun, existing) {
+			candidatesByRepo[repoKey] = pipelineRun
+		}
+	}
+
+	selected := make([]*tektonv1.PipelineRun, 0, len(candidatesByRepo))
+	for _, pipelineRun := range candidatesByRepo {
+		selected = append(selected, pipelineRun)
+	}
+
+	sort.Slice(selected, func(i, j int) bool {
+		return shouldPreferLeaseQueueRecoveryCandidate(selected[i], selected[j])
+	})
+
+	recoveryKeys := make([]types.NamespacedName, 0, len(selected))
+	for _, pipelineRun := range selected {
+		recoveryKeys = append(recoveryKeys, types.NamespacedName{
+			Namespace: pipelineRun.GetNamespace(),
+			Name:      pipelineRun.GetName(),
+		})
+	}
+	return recoveryKeys
+}
+
+func isEligibleLeaseQueueRecoveryCandidate(pipelineRun *tektonv1.PipelineRun) bool {
+	if pipelineRun == nil {
+		return false
+	}
+	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"
+}
+
+func shouldPreferLeaseQueueRecoveryCandidate(left, right *tektonv1.PipelineRun) bool {
+	if left.CreationTimestamp.Equal(&right.CreationTimestamp) {
+		return left.GetName() < right.GetName()
+	}
+	return left.CreationTimestamp.Before(&right.CreationTimestamp)
+}
+
 // enqueue only the pipelineruns which are in `started` state
 // pipelinerun will have a label `pipelinesascode.tekton.dev/state` to describe the state.
 func checkStateAndEnqueue(impl *controller.Impl) func(obj any) {

pkg/reconciler/controller_test.go

@@ -4,15 +4,20 @@ import (
 	"context"
 	"path"
 	"testing"
+	"time"
 
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/keys"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/kubeinteraction"
 	tektontest "github.com/openshift-pipelines/pipelines-as-code/pkg/test/tekton"
 	pipelinev1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1"
+	tektonv1lister "github.com/tektoncd/pipeline/pkg/client/listers/pipeline/v1"
 	"go.uber.org/zap"
 	zapobserver "go.uber.org/zap/zaptest/observer"
 	"gotest.tools/v3/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 )
 
@@ -81,3 +86,125 @@ func TestCtrlOpts(t *testing.T) {
 	// Assert that the promote filter function returns true.
 	assert.Assert(t, promote)
 }
+
+func TestSelectLeaseQueueRecoveryKeys(t *testing.T) {
+	now := time.Unix(1_700_001_000, 0)
+
+	tests := []struct {
+		name         string
+		pipelineRuns []*pipelinev1.PipelineRun
+		want         []string
+	}{
+		{
+			name: "selects one oldest queued pending run per repository",
+			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",
+				}, false),
+				newLeaseRecoveryPR("valid", "test-ns", "repo-b", now.Add(3*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)),
+				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",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			keys := selectLeaseQueueRecoveryKeys(tt.pipelineRuns)
+			got := make([]string, 0, len(keys))
+			for _, key := range keys {
+				got = append(got, key.String())
+			}
+			assert.DeepEqual(t, got, tt.want)
+		})
+	}
+}
+
+func TestRunLeaseQueueRecovery(t *testing.T) {
+	observer, catcher := zapobserver.New(zap.DebugLevel)
+	logger := zap.New(observer).Sugar()
+	wh := &fakeReconciler{}
+	impl := controller.NewContext(context.TODO(), wh, controller.ControllerOptions{
+		WorkQueueName: "LeaseRecovery",
+		Logger:        logger.Named("LeaseRecovery"),
+	})
+
+	now := time.Unix(1_700_001_100, 0)
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+	for _, pipelineRun := range []*pipelinev1.PipelineRun{
+		newLeaseRecoveryPR("first", "test-ns", "repo-a", now, map[string]string{}, false),
+		newLeaseRecoveryPR("second", "test-ns", "repo-a", now.Add(time.Second), map[string]string{}, false),
+		newLeaseRecoveryPR("third", "test-ns", "repo-b", now.Add(2*time.Second), map[string]string{}, false),
+		newLeaseRecoveryStartedPR("started", "test-ns", "repo-c", now.Add(3*time.Second)),
+	} {
+		assert.NilError(t, indexer.Add(pipelineRun))
+	}
+
+	runLeaseQueueRecovery(logger, impl, tektonv1lister.NewPipelineRunLister(indexer))
+
+	assert.Equal(t, catcher.FilterMessageSnippet("Adding to queue test-ns/first").Len(), 1)
+	assert.Equal(t, catcher.FilterMessageSnippet("Adding to queue test-ns/third").Len(), 1)
+	assert.Equal(t, catcher.FilterMessageSnippet("Adding to queue").Len(), 2)
+}
+
+func newLeaseRecoveryPR(
+	name, namespace, repo string,
+	createdAt time.Time,
+	extraAnnotations map[string]string,
+	done bool,
+) *pipelinev1.PipelineRun {
+	annotations := map[string]string{
+		keys.State: kubeinteraction.StateQueued,
+	}
+	if repo != "" {
+		annotations[keys.Repository] = repo
+	}
+	for key, value := range extraAnnotations {
+		annotations[key] = value
+	}
+
+	pipelineRun := &pipelinev1.PipelineRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              name,
+			Namespace:         namespace,
+			CreationTimestamp: metav1.Time{Time: createdAt},
+			Annotations:       annotations,
+		},
+		Spec: pipelinev1.PipelineRunSpec{
+			Status: pipelinev1.PipelineRunSpecStatusPending,
+		},
+	}
+	if done {
+		pipelineRun.Status.Conditions = duckv1.Conditions{{
+			Type:   "Succeeded",
+			Status: "True",
+		}}
+	}
+	return pipelineRun
+}
+
+func newLeaseRecoveryStartedPR(name, namespace, repo string, createdAt time.Time) *pipelinev1.PipelineRun {
+	return &pipelinev1.PipelineRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              name,
+			Namespace:         namespace,
+			CreationTimestamp: metav1.Time{Time: createdAt},
+			Annotations: map[string]string{
+				keys.Repository: repo,
+				keys.State:      kubeinteraction.StateStarted,
+			},
+		},
+		Spec: pipelinev1.PipelineRunSpec{
+			Status: pipelinev1.PipelineRunSpecStatusPending,
+		},
+	}
+}