fix(queue): guard malformed execution-order annotations

pkg/queue/queue_manager.go

@@ -17,6 +17,7 @@ import (
 	"go.uber.org/zap"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/pkg/logging"
 )
 
 const (
@@ -181,8 +182,16 @@ func (qm *Manager) RemoveAndTakeItemFromQueue(repo *v1alpha1.Repository, run *te
 func FilterPipelineRunByState(ctx context.Context, tekton versioned2.Interface, orderList []string, wantedStatus, wantedState string) []string {
 	orderedList := []string{}
 	for _, prName := range orderList {
-		prKey := strings.Split(prName, "/")
-		pr, err := tekton.TektonV1().PipelineRuns(prKey[0]).Get(ctx, prKey[1], v1.GetOptions{})
+		// The list comes straight from a user-editable annotation, so a
+		// malformed entry must be skipped, not indexed: a panic here takes
+		// down the whole watcher, and from InitQueues it does so on every
+		// restart.
+		namespace, name, ok := SplitPrKey(prName)
+		if !ok {
+			logging.FromContext(ctx).Warnf("ignoring malformed execution-order entry %q", prName)
+			continue
+		}
+		pr, err := tekton.TektonV1().PipelineRuns(namespace).Get(ctx, name, v1.GetOptions{})
 		if err != nil {
 			continue
 		}
@@ -196,7 +205,7 @@ func FilterPipelineRunByState(ctx context.Context, tekton versioned2.Interface,
 			if wantedStatus != "" && pr.Spec.Status != tektonv1.PipelineRunSpecStatus(wantedStatus) {
 				continue
 			}
-			orderedList = append(orderedList, prName)
+			orderedList = append(orderedList, namespace+"/"+name)
 		}
 	}
 	return orderedList

pkg/queue/queue_manager_interface.go

@@ -3,6 +3,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/v1alpha1"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/generated/clientset/versioned"
@@ -28,3 +29,21 @@ func RepoKey(repo *v1alpha1.Repository) string {
 func PrKey(run *tektonv1.PipelineRun) string {
 	return fmt.Sprintf("%s/%s", run.Namespace, run.Name)
 }
+
+// SplitPrKey parses a "namespace/name" queue key, the inverse of PrKey.
+//
+// The key can come straight from a user-editable annotation (execution-order)
+// or from an internal queue that in principle only ever stores what PrKey
+// produced, but every caller needs the same defensive parsing: a malformed
+// entry must be rejected, not indexed, since callers use namespace/name to
+// index directly into a slice or make a Tekton client call, and a panic here
+// has in the past taken down the whole watcher on every restart.
+func SplitPrKey(key string) (namespace, name string, ok bool) {
+	namespace, name, found := strings.Cut(strings.TrimSpace(key), "/")
+	namespace = strings.TrimSpace(namespace)
+	name = strings.TrimSpace(name)
+	if !found || namespace == "" || name == "" || strings.Contains(name, "/") {
+		return "", "", false
+	}
+	return namespace, name, true
+}

pkg/queue/queue_manager_interface_test.go

@@ -0,0 +1,35 @@
+package queue
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+func TestSplitPrKey(t *testing.T) {
+	tests := []struct {
+		name          string
+		key           string
+		wantNamespace string
+		wantName      string
+		wantOK        bool
+	}{
+		{name: "valid key", key: "ns/name", wantNamespace: "ns", wantName: "name", wantOK: true},
+		{name: "trims surrounding and inner whitespace", key: "  ns / name  ", wantNamespace: "ns", wantName: "name", wantOK: true},
+		{name: "empty string", key: "", wantOK: false},
+		{name: "missing separator", key: "no-slash", wantOK: false},
+		{name: "empty namespace", key: "/name-only", wantOK: false},
+		{name: "empty name", key: "ns-only/", wantOK: false},
+		{name: "name contains a slash", key: "too/many/slashes", wantOK: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			namespace, name, ok := SplitPrKey(tt.key)
+			assert.Equal(t, ok, tt.wantOK)
+			if tt.wantOK {
+				assert.Equal(t, namespace, tt.wantNamespace)
+				assert.Equal(t, name, tt.wantName)
+			}
+		})
+	}
+}

pkg/queue/queue_manager_test.go

@@ -328,6 +328,47 @@ func TestFilterPipelineRunByInProgress(t *testing.T) {
 	assert.DeepEqual(t, filtered, expected)
 }
 
+// TestFilterPipelineRunByStateSkipsMalformedKeys asserts that a malformed
+// entry in the execution-order annotation is skipped rather than panicking.
+// The annotation is user-editable, and this function runs inside InitQueues at
+// watcher startup, so a panic here is a persistent CrashLoopBackOff for the
+// whole cluster, not one dropped reconcile.
+func TestFilterPipelineRunByStateSkipsMalformedKeys(t *testing.T) {
+	ctx, _ := rtesting.SetupFakeContext(t)
+	ns := "test-ns"
+
+	pipelineRuns := []*tektonv1.PipelineRun{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "valid",
+				Namespace: ns,
+				Annotations: map[string]string{
+					keys.State: kubeinteraction.StateQueued,
+				},
+			},
+			Spec: tektonv1.PipelineRunSpec{
+				Status: tektonv1.PipelineRunSpecStatusPending,
+			},
+		},
+	}
+	stdata, _ := testclient.SeedTestData(t, ctx, testclient.Data{
+		Namespaces:   []*corev1.Namespace{{ObjectMeta: metav1.ObjectMeta{Name: ns}}},
+		PipelineRuns: pipelineRuns,
+	})
+
+	orderList := []string{
+		"",                 // strings.Split("", ",") yields this
+		"no-slash",         // missing namespace separator
+		"/name-only",       // empty namespace
+		"ns-only/",         // empty name
+		"too/many/slashes", // name would contain a slash
+		"  " + ns + " / valid  ",
+		ns + "/valid",
+	}
+	filtered := FilterPipelineRunByState(ctx, stdata.Pipeline, orderList, tektonv1.PipelineRunSpecStatusPending, kubeinteraction.StateQueued)
+	assert.DeepEqual(t, filtered, []string{ns + "/valid", ns + "/valid"})
+}
+
 // TestQueueManagerInitQueuesSkipsPipelineRunsWithoutOrder asserts that a
 // PipelineRun without an execution-order annotation is skipped rather than
 // aborting the whole initialization. The order-less runs are deliberately the

pkg/reconciler/finalizer.go

@@ -3,7 +3,6 @@ package reconciler
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/keys"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/v1alpha1"
@@ -72,20 +71,12 @@ func (r *Reconciler) finalizeKind(ctx context.Context, pr *tektonv1.PipelineRun)
 			}
 		}
 		logger = logger.With("namespace", repo.Namespace)
-		next := r.qm.RemoveAndTakeItemFromQueue(repo, pr)
-		if next != "" {
-			key := strings.Split(next, "/")
-			pr, err := r.run.Clients.Tekton.TektonV1().PipelineRuns(key[0]).Get(ctx, key[1], metav1.GetOptions{})
-			if err != nil {
-				return err
-			}
-			if err := r.
-				updatePipelineRunToInProgress(ctx, logger, repo, pr); err != nil {
-				logger.Errorf("failed to update status: %w", err)
-				return err
-			}
-			return nil
-		}
+		// This releases the slot the deleted PipelineRun held and promotes the
+		// next candidate, retrying past a malformed or already-gone queue key
+		// instead of leaving it stuck in the running set forever with no real
+		// PipelineRun ever completing to free it.
+		r.startNextPipelineRunInQueue(ctx, logger, repo, pr)
+		return nil
 	}
 	return nil
 }

pkg/reconciler/finalizer_test.go

@@ -371,3 +371,149 @@ func TestReconcilerFinalizeKind(t *testing.T) {
 		})
 	}
 }
+
+// TestReconcilerFinalizeKindSkipsMalformedQueueKey is a regression test for a
+// panic on a malformed key taken off the queue: FinalizeKind splits and
+// indexes the "next" key returned by RemoveAndTakeItemFromQueue the same way
+// queue_pipelineruns.go and queue_manager.go do for execution-order entries,
+// so it needs the same guard against an entry that doesn't split into exactly
+// namespace/name. A malformed queue entry is not reachable through normal
+// annotation parsing today (see TestFilterPipelineRunByStateSkipsMalformedKeys),
+// but the guard here is deliberately defensive rather than relying on that
+// invariant holding everywhere a string is added to the queue.
+func TestReconcilerFinalizeKindSkipsMalformedQueueKey(t *testing.T) {
+	observer, logs := zapobserver.New(zap.InfoLevel)
+	fakelogger := zap.New(observer).Sugar()
+
+	ctx, _ := rtesting.SetupFakeContext(t)
+	ctx = logging.WithLogger(ctx, fakelogger)
+
+	stdata, informers := testclient.SeedTestData(t, ctx, testclient.Data{
+		Repositories: []*v1alpha1.Repository{finalizeTestRepo},
+	})
+
+	cs := &params.Run{
+		Clients: clients.Clients{
+			PipelineAsCode: stdata.PipelineAsCode,
+			Kube:           stdata.Kube,
+			Log:            fakelogger,
+		},
+		Info: info.Info{
+			Kube:       &info.KubeOpts{Namespace: "pac"},
+			Controller: &info.ControllerInfo{GlobalRepository: "pac"},
+			Pac:        info.NewPacOpts(),
+		},
+	}
+	cs.Clients.SetConsoleUI(consoleui.FallBackConsole{})
+	r := Reconciler{
+		repoLister: informers.Repository.Lister(),
+		qm:         queuepkg.NewManager(fakelogger),
+		run:        cs,
+		kinteract:  &testkubernetestint.KinterfaceTest{},
+	}
+
+	running := getTestPR("running", kubeinteraction.StateStarted)
+	_, err := r.qm.AddListToRunningQueue(finalizeTestRepo, []string{queuepkg.PrKey(running)})
+	assert.NilError(t, err)
+	// A malformed entry, as if something other than PrKey had added it to the
+	// queue: no "/" separator to split on namespace/name.
+	assert.NilError(t, r.qm.AddToPendingQueue(finalizeTestRepo, []string{"malformed-entry-with-no-slash"}))
+
+	err = r.FinalizeKind(ctx, running)
+	assert.NilError(t, err, "a malformed queue entry must not panic or fail the finalizer")
+
+	found := false
+	for _, entry := range logs.All() {
+		if strings.Contains(entry.Message, "invalid pipelineRun key") {
+			found = true
+		}
+	}
+	assert.Assert(t, found, "expected a warning about the malformed queue entry, got: %v", logs.All())
+}
+
+// TestReconcilerFinalizeKindPromotesSuccessorPastMalformedKey is a regression
+// test for a slot leak: RemoveAndTakeItemFromQueue already moves the "next"
+// key into the running set before FinalizeKind gets a chance to validate it,
+// so simply discarding a malformed key without releasing that reservation
+// would strand it forever, since nothing ever completes to release a slot
+// that was never a real running PipelineRun. FinalizeKind must keep retrying
+// past the malformed entry and promote the next valid candidate instead of
+// leaving the running queue with a phantom occupant.
+func TestReconcilerFinalizeKindPromotesSuccessorPastMalformedKey(t *testing.T) {
+	observer, logs := zapobserver.New(zap.InfoLevel)
+	fakelogger := zap.New(observer).Sugar()
+
+	ctx, _ := rtesting.SetupFakeContext(t)
+	ctx = logging.WithLogger(ctx, fakelogger)
+
+	_, mux, mockServerURL, teardown := ghtesthelper.SetupGH()
+	defer teardown()
+	mux.HandleFunc("/repos/org/repo/statuses/123afc", func(rw http.ResponseWriter, _ *http.Request) {
+		fmt.Fprint(rw, `{"state":"pending"}`)
+	})
+
+	repo := finalizeTestRepo.DeepCopy()
+	repo.Spec.GitProvider.URL = mockServerURL
+
+	successor := getTestPR("successor", kubeinteraction.StateQueued)
+
+	stdata, informers := testclient.SeedTestData(t, ctx, testclient.Data{
+		Repositories: []*v1alpha1.Repository{repo},
+		PipelineRuns: []*tektonv1.PipelineRun{successor},
+		ConfigMap: []*corev1.ConfigMap{{
+			ObjectMeta: metav1.ObjectMeta{Name: "pipelines-as-code", Namespace: system.Namespace()},
+			Data: map[string]string{
+				settings.TrustedProviderHostnamesKey: strings.TrimPrefix(mockServerURL, "http://"),
+			},
+		}},
+	})
+
+	cs := &params.Run{
+		Clients: clients.Clients{
+			PipelineAsCode: stdata.PipelineAsCode,
+			Kube:           stdata.Kube,
+			Tekton:         stdata.Pipeline,
+			Log:            fakelogger,
+		},
+		Info: info.Info{
+			Kube:       &info.KubeOpts{Namespace: "pac"},
+			Controller: &info.ControllerInfo{GlobalRepository: "pac"},
+			Pac:        info.NewPacOpts(),
+		},
+	}
+	cs.Clients.SetConsoleUI(consoleui.FallBackConsole{})
+	r := Reconciler{
+		repoLister: informers.Repository.Lister(),
+		qm:         queuepkg.NewManager(fakelogger),
+		run:        cs,
+		kinteract: &testkubernetestint.KinterfaceTest{
+			GetSecretResult: map[string]string{
+				"pac-git-basic-auth-owner-repo": "https://whateveryousayboss",
+			},
+		},
+	}
+
+	running := getTestPR("running", kubeinteraction.StateStarted)
+	_, err := r.qm.AddListToRunningQueue(repo, []string{queuepkg.PrKey(running)})
+	assert.NilError(t, err)
+	// The malformed entry is queued ahead of the valid successor so
+	// RemoveAndTakeItemFromQueue picks it first and FinalizeKind has to skip
+	// past it rather than stopping there.
+	assert.NilError(t, r.qm.AddToPendingQueue(repo, []string{"malformed-entry-with-no-slash"}))
+	assert.NilError(t, r.qm.AddToPendingQueue(repo, []string{queuepkg.PrKey(successor)}))
+
+	err = r.FinalizeKind(ctx, running)
+	assert.NilError(t, err, "a malformed queue entry must not panic or fail the finalizer")
+
+	found := false
+	for _, entry := range logs.All() {
+		if strings.Contains(entry.Message, "invalid pipelineRun key") {
+			found = true
+		}
+	}
+	assert.Assert(t, found, "expected a warning about the malformed queue entry, got: %v", logs.All())
+
+	runningKeys := r.qm.RunningPipelineRuns(repo)
+	assert.DeepEqual(t, runningKeys, []string{queuepkg.PrKey(successor)})
+	assert.Equal(t, len(r.qm.QueuedPipelineRuns(repo)), 0, "the successor must have been promoted, not left pending")
+}

pkg/reconciler/queue_pipelineruns.go

@@ -85,14 +85,14 @@ func (r *Reconciler) queuePipelineRun(ctx context.Context, logger *zap.SugaredLo
 		dropped := map[string]bool{}
 		for _, prKeys := range acquired {
 			repoKey := queuepkg.RepoKey(repo)
-			nsName := strings.Split(prKeys, "/")
-			if len(nsName) != 2 {
+			namespace, name, ok := queuepkg.SplitPrKey(prKeys)
+			if !ok {
 				logger.Errorf("invalid pipelineRun key %q queued for repository %s, dropping it", prKeys, repo.GetName())
 				_ = r.qm.RemoveFromQueue(repoKey, prKeys)
 				dropped[prKeys] = true
 				continue
 			}
-			acquiredPR, err := r.run.Clients.Tekton.TektonV1().PipelineRuns(nsName[0]).Get(ctx, nsName[1], metav1.GetOptions{})
+			acquiredPR, err := r.run.Clients.Tekton.TektonV1().PipelineRuns(namespace).Get(ctx, name, metav1.GetOptions{})
 			if err != nil {
 				// Nothing has been written to the cluster yet, so this PipelineRun
 				// is still pending and cannot be running. Hand the slot back: if we