feat: Add k8 Lease backend for concurrency coord

Introduced an opt-in concurrency backend that used Kubernetes Lease
objects and PipelineRun annotations for queue coordination. This
addressed potential state drift and race conditions during watcher
restarts or API delays by storing the queue state in the cluster
instead of only in memory. Added a global configuration setting to
choose between the legacy memory backend and the new lease-based
logic. Updated the queue manager interface to support context-aware
operations and ensured that stale claims were automatically
reclaimed via TTL-based expiration.

The system preserved existing lease objects during the release process
instead of completely deleting them. This reused the same records
across multiple acquisition cycles, which reduced excessive
communication overhead with the cluster.

AI-assisted-by: Cursor (Codex)
Signed-off-by: Chmouel Boudjnah <chmouel@redhat.com>

Author: chmouel

config/302-pac-configmap.yaml

@@ -173,6 +173,13 @@ data:
   # Default: true
   skip-push-event-for-pr-commits: "true"
 
+  # Selects the concurrency queue backend used by the watcher.
+  # "memory" keeps the existing in-process queue state.
+  # "lease" uses Kubernetes Leases plus PipelineRun claims to recover more safely
+  # from watcher restarts and cluster/API timing issues.
+  # Restart the watcher after changing this setting.
+  concurrency-backend: "memory"
+
   # Configure a custom console here, the driver support custom parameters from
   # Repo CR along a few other template variable, see documentation for more
   # details

docs/content/docs/advanced/concurrency.md

@@ -4,6 +4,11 @@ weight: 2
 ---
 This page illustrates how Pipelines-as-Code manages concurrent PipelineRun execution. When you set a concurrency limit on a Repository CR, Pipelines-as-Code queues incoming PipelineRuns and starts them only when capacity allows.
 
+The watcher supports two queue backends controlled by the global `concurrency-backend` setting in the `pipelines-as-code` ConfigMap:
+
+- `memory` keeps queue state in the watcher process. This is the historical behavior and remains the default.
+- `lease` stores queue coordination in Kubernetes using `Lease` objects and short-lived PipelineRun claims. This mode is more resilient when the watcher restarts or the cluster is slow to reconcile updates.
+
 ## Flow diagram
 
 ```mermaid
@@ -37,3 +42,59 @@ graph TD
     K --> |No| N
 
 ```
+
+## Backend selection
+
+To enable the Kubernetes-backed queue coordination, set the global config to:
+
+```yaml
+data:
+  concurrency-backend: "lease"
+```
+
+Restart the watcher after changing `concurrency-backend`; the backend is selected at startup.
+
+When `lease` mode is enabled, Pipelines-as-Code still uses the existing `queued`, `started`, and `completed` PipelineRun states. The difference is that promotion of the next queued PipelineRun is serialized with a per-repository `Lease`, which reduces queue drift during cluster/API instability.
+
+## Debugging the Lease Backend
+
+When `concurrency-backend: "lease"` is enabled, queued `PipelineRun`s expose queue debugging state directly in annotations:
+
+- `pipelinesascode.tekton.dev/queue-decision`
+- `pipelinesascode.tekton.dev/queue-debug-summary`
+- `pipelinesascode.tekton.dev/queue-claimed-by`
+- `pipelinesascode.tekton.dev/queue-claimed-at`
+- `pipelinesascode.tekton.dev/queue-promotion-retries`
+- `pipelinesascode.tekton.dev/queue-promotion-last-error`
+
+This makes it possible to diagnose most queue issues with `kubectl` before looking at watcher logs.
+
+### Useful commands
+
+```bash
+kubectl get pipelinerun -n <namespace> <name> -o jsonpath='{.metadata.annotations.pipelinesascode\.tekton\.dev/queue-decision}{"\n"}'
+kubectl get pipelinerun -n <namespace> <name> -o jsonpath='{.metadata.annotations.pipelinesascode\.tekton\.dev/queue-debug-summary}{"\n"}'
+kubectl describe pipelinerun -n <namespace> <name>
+kubectl get events -n <namespace> --field-selector involvedObject.kind=Repository
+```
+
+### Queue decisions
+
+- `waiting_for_slot`: the run is queued and waiting for repository capacity.
+- `claim_active`: another watcher already holds an active short-lived claim on this run.
+- `claimed_for_promotion`: this run has been claimed and is being promoted to `started`.
+- `promotion_failed`: the watcher failed while promoting the run to `started`.
+- `recovery_requeued`: the lease recovery loop noticed this run and enqueued it again.
+- `missing_execution_order`: the run is queued but its execution order annotation does not include itself.
+- `not_recoverable`: the run is still `queued` but is no longer eligible for lease recovery.
+
+### Events
+
+The watcher also emits repository-scoped Kubernetes events for the most important transitions:
+
+- `QueueClaimedForPromotion`
+- `QueuePromotionFailed`
+- `QueueRecoveryRequeued`
+- `QueueLeaseAcquireTimeout`
+
+If the queue decision and events do not explain the behavior, switch the watcher to debug logging and grep for the repository key and PipelineRun key. The lease backend logs include lease acquisition attempts, active claim evaluation, queue-state snapshots, and recovery loop selections.

docs/content/docs/api/configmap.md

@@ -335,6 +335,20 @@ skip-push-event-for-pr-commits: "true"
 
 {{< /param >}}
 
+{{< param name="concurrency-backend" type="string" default="memory" id="param-concurrency-backend" >}}
+Selects the queue coordination backend used by the watcher. Supported values:
+
+- `memory`: in-process queue tracking. This is the default and matches the historical behavior.
+- `lease`: Kubernetes-backed coordination using `Lease` objects and short-lived PipelineRun claims for improved recovery during watcher restarts or API instability.
+
+Restart the watcher after changing this setting.
+
+```yaml
+concurrency-backend: "memory"
+```
+
+{{< /param >}}
+
 ## Complete Example
 
 ```yaml
@@ -371,6 +385,7 @@ data:
   remember-ok-to-test: "true"
   require-ok-to-test-sha: "false"
   skip-push-event-for-pr-commits: "true"
+  concurrency-backend: "memory"
 ```
 
 ## Updating configuration

hack/gh-workflow-ci.sh

@@ -21,6 +21,9 @@ create_pac_github_app_secret() {
   kubectl patch configmap -n pipelines-as-code -p "{\"data\":{\"bitbucket-cloud-check-source-ip\": \"false\"}}" \
     --type merge pipelines-as-code
 
+  # Enable lease backend concurrency
+  kubectl patch configmap -n pipelines-as-code pipelines-as-code -p '{"data":{"concurrency-backend":"lease"}}'
+
   # restart controller
   kubectl -n pipelines-as-code delete pod -l app.kubernetes.io/name=controller
 

pkg/apis/pipelinesascode/keys/keys.go

@@ -61,6 +61,13 @@ const (
 	LogURL                 = pipelinesascode.GroupName + "/log-url"
 	ExecutionOrder         = pipelinesascode.GroupName + "/execution-order"
 	SCMReportingPLRStarted = pipelinesascode.GroupName + "/scm-reporting-plr-started"
+	QueueClaimedBy         = pipelinesascode.GroupName + "/queue-claimed-by"
+	QueueClaimedAt         = pipelinesascode.GroupName + "/queue-claimed-at"
+	QueueDecision          = pipelinesascode.GroupName + "/queue-decision"
+	QueueDebugSummary      = pipelinesascode.GroupName + "/queue-debug-summary"
+	QueuePromotionRetries  = pipelinesascode.GroupName + "/queue-promotion-retries"
+	QueuePromotionBlocked  = pipelinesascode.GroupName + "/queue-promotion-blocked"
+	QueuePromotionLastErr  = pipelinesascode.GroupName + "/queue-promotion-last-error"
 	SecretCreated          = pipelinesascode.GroupName + "/secret-created"
 	CloneURL               = pipelinesascode.GroupName + "/clone-url"
 	// PublicGithubAPIURL default is "https://api.github.com" but it can be overridden by X-GitHub-Enterprise-Host header.

pkg/params/config_sync.go

@@ -13,6 +13,10 @@ import (
 	"knative.dev/pkg/system"
 )
 
+var terminateProcessForConfigChange = func() {
+	_ = syscall.Kill(os.Getpid(), syscall.SIGTERM)
+}
+
 func StartConfigSync(ctx context.Context, run *Run) {
 	// init pac config
 	_ = run.UpdatePacConfig(ctx)
@@ -28,7 +32,19 @@ func StartConfigSync(ctx context.Context, run *Run) {
 			// nothing to do
 		},
 		UpdateFunc: func(_, _ any) {
-			_ = run.UpdatePacConfig(ctx)
+			oldBackend, newBackend, changed, err := updatePacConfigAndDetectBackendChange(ctx, run)
+			if err != nil {
+				return
+			}
+			if changed {
+				if run.Clients.Log != nil {
+					run.Clients.Log.Infof(
+						"concurrency-backend changed from %q to %q; restarting process so the queue backend is recreated",
+						oldBackend, newBackend,
+					)
+				}
+				terminateProcessForConfigChange()
+			}
 		},
 		DeleteFunc: func(_ any) {
 			// nothing to do
@@ -46,3 +62,13 @@ func StartConfigSync(ctx context.Context, run *Run) {
 	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
 	<-sig
 }
+
+func updatePacConfigAndDetectBackendChange(ctx context.Context, run *Run) (string, string, bool, error) {
+	oldBackend := run.Info.GetPacOpts().ConcurrencyBackend
+	if err := run.UpdatePacConfig(ctx); err != nil {
+		return oldBackend, oldBackend, false, err
+	}
+
+	newBackend := run.Info.GetPacOpts().ConcurrencyBackend
+	return oldBackend, newBackend, oldBackend != "" && newBackend != "" && oldBackend != newBackend, nil
+}

pkg/params/config_sync_test.go

@@ -0,0 +1,89 @@
+package params
+
+import (
+	"testing"
+
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/consoleui"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/clients"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/info"
+	"github.com/openshift-pipelines/pipelines-as-code/pkg/params/settings"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubefake "k8s.io/client-go/kubernetes/fake"
+	rtesting "knative.dev/pkg/reconciler/testing"
+
+	"go.uber.org/zap"
+	"gotest.tools/v3/assert"
+)
+
+func TestUpdatePacConfigAndDetectBackendChange(t *testing.T) {
+	tests := []struct {
+		name           string
+		initialBackend string
+		configData     map[string]string
+		wantOld        string
+		wantNew        string
+		wantChanged    bool
+	}{
+		{
+			name:           "detects backend change",
+			initialBackend: settings.ConcurrencyBackendMemory,
+			configData: map[string]string{
+				"concurrency-backend":  settings.ConcurrencyBackendLease,
+				"tekton-dashboard-url": "https://dashboard.example.test",
+			},
+			wantOld:     settings.ConcurrencyBackendMemory,
+			wantNew:     settings.ConcurrencyBackendLease,
+			wantChanged: true,
+		},
+		{
+			name:           "ignores unchanged backend",
+			initialBackend: settings.ConcurrencyBackendLease,
+			configData: map[string]string{
+				"concurrency-backend":  settings.ConcurrencyBackendLease,
+				"tekton-dashboard-url": "https://dashboard.example.test",
+			},
+			wantOld:     settings.ConcurrencyBackendLease,
+			wantNew:     settings.ConcurrencyBackendLease,
+			wantChanged: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, _ := rtesting.SetupFakeContext(t)
+			ctx = info.StoreNS(ctx, "pac")
+
+			run := &Run{
+				Clients: clients.Clients{
+					Kube: kubefake.NewSimpleClientset(&corev1.ConfigMap{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "pac-config",
+							Namespace: "pac",
+						},
+						Data: tt.configData,
+					}),
+					Log: zap.NewNop().Sugar(),
+				},
+				Info: info.Info{
+					Pac: &info.PacOpts{
+						Settings: settings.Settings{
+							ConcurrencyBackend: tt.initialBackend,
+						},
+					},
+					Controller: &info.ControllerInfo{
+						Configmap: "pac-config",
+					},
+				},
+			}
+			run.Clients.InitClients()
+			run.Clients.SetConsoleUI(consoleui.FallBackConsole{})
+
+			oldBackend, newBackend, changed, err := updatePacConfigAndDetectBackendChange(ctx, run)
+			assert.NilError(t, err)
+			assert.Equal(t, oldBackend, tt.wantOld)
+			assert.Equal(t, newBackend, tt.wantNew)
+			assert.Equal(t, changed, tt.wantChanged)
+		})
+	}
+}

pkg/params/settings/config.go

@@ -15,6 +15,8 @@ import (
 
 const (
 	PACApplicationNameDefaultValue = "Pipelines as Code CI"
+	ConcurrencyBackendMemory       = "memory"
+	ConcurrencyBackendLease        = "lease"
 
 	HubURLKey                          = "hub-url"
 	HubCatalogNameKey                  = "hub-catalog-name"
@@ -80,8 +82,9 @@ type Settings struct {
 	CustomConsolePRTaskLog    string `json:"custom-console-url-pr-tasklog"`
 	CustomConsoleNamespaceURL string `json:"custom-console-url-namespace"`
 
-	RememberOKToTest   bool `json:"remember-ok-to-test"`
-	RequireOkToTestSHA bool `json:"require-ok-to-test-sha"`
+	RememberOKToTest   bool   `json:"remember-ok-to-test"`
+	RequireOkToTestSHA bool   `json:"require-ok-to-test-sha"`
+	ConcurrencyBackend string `default:"memory"              json:"concurrency-backend"`
 }
 
 func (s *Settings) DeepCopy(out *Settings) {
@@ -110,6 +113,7 @@ func DefaultValidators() map[string]func(string) error {
 		"CustomConsoleURL":           isValidURL,
 		"CustomConsolePRTaskLog":     startWithHTTPorHTTPS,
 		"CustomConsolePRDetail":      startWithHTTPorHTTPS,
+		"ConcurrencyBackend":         isValidConcurrencyBackend,
 	}
 }
 
@@ -159,3 +163,12 @@ func startWithHTTPorHTTPS(url string) error {
 	}
 	return nil
 }
+
+func isValidConcurrencyBackend(backend string) error {
+	switch backend {
+	case "", ConcurrencyBackendMemory, ConcurrencyBackendLease:
+		return nil
+	default:
+		return fmt.Errorf("invalid concurrency backend %q, must be one of %q or %q", backend, ConcurrencyBackendMemory, ConcurrencyBackendLease)
+	}
+}

pkg/params/settings/config_test.go

@@ -49,6 +49,7 @@ func TestSyncConfig(t *testing.T) {
 				CustomConsolePRTaskLog:               "",
 				CustomConsoleNamespaceURL:            "",
 				RememberOKToTest:                     false,
+				ConcurrencyBackend:                   ConcurrencyBackendMemory,
 			},
 		},
 		{
@@ -79,6 +80,7 @@ func TestSyncConfig(t *testing.T) {
 				"remember-ok-to-test":                     "false",
 				"skip-push-event-for-pr-commits":          "true",
 				"require-ok-to-test-sha":                  "true",
+				"concurrency-backend":                     "lease",
 			},
 			expectedStruct: Settings{
 				ApplicationName:                      "pac-pac",
@@ -110,6 +112,7 @@ func TestSyncConfig(t *testing.T) {
 				CustomConsoleNamespaceURL:            "https://custom-console-namespace",
 				RememberOKToTest:                     false,
 				RequireOkToTestSHA:                   true,
+				ConcurrencyBackend:                   ConcurrencyBackendLease,
 			},
 		},
 		{
@@ -147,6 +150,13 @@ func TestSyncConfig(t *testing.T) {
 			},
 			expectedError: "custom validation failed for field CustomConsolePRTaskLog: invalid value, must start with http:// or https://",
 		},
+		{
+			name: "invalid concurrency backend",
+			configMap: map[string]string{
+				"concurrency-backend": "sqlite",
+			},
+			expectedError: `custom validation failed for field ConcurrencyBackend: invalid concurrency backend "sqlite", must be one of "memory" or "lease"`,
+		},
 	}
 
 	for _, tc := range testCases {