fix: Annotate PLR when when started status is reported

The use of the state label (which is mutable) for deciding when to report
to the SCM that the PLR was started is flaky.

It was seen that the reconciler get events about PLRs with unexpected value
for the state label. For example, after the status is reported to the SCM,
and the label value is patched to "started", after serval reconcile iterations
the label had the "queued" value again. This can happen because of unsafe patching
done by controllers (not just the PAC controllers) which reconciles PLRs.

Introduce a new annotation for indicating the the status was reported to the SCM.
By adding an annotation which is set once, we remove the risk that its value will
get overwritten by other controllers (since maps are merged when patched, values
are not getting removed unless explicitly defined in the patch
- https://datatracker.ietf.org/doc/html/rfc7386#section-2)

In addition, at the start of each reconcile loop, ensure that we operate on the latest
version of the PLR and not using a stale value from the cache.

Signed-off-by: Gal Ben Haim <[email protected]>

Author: gbenhaim

pkg/apis/pipelinesascode/keys/keys.go

@@ -61,10 +61,11 @@ const (
 	LogURL             = pipelinesascode.GroupName + "/log-url"
 	ExecutionOrder     = pipelinesascode.GroupName + "/execution-order"
 	// PublicGithubAPIURL default is "https://api.github.com" but it can be overridden by X-GitHub-Enterprise-Host header.
-	PublicGithubAPIURL   = "https://api.github.com"
-	GithubApplicationID  = "github-application-id"
-	GithubPrivateKey     = "github-private-key"
-	ResultsRecordSummary = "results.tekton.dev/recordSummaryAnnotations"
+	PublicGithubAPIURL     = "https://api.github.com"
+	GithubApplicationID    = "github-application-id"
+	GithubPrivateKey       = "github-private-key"
+	ResultsRecordSummary   = "results.tekton.dev/recordSummaryAnnotations"
+	SCMReportingPLRStarted = pipelinesascode.GroupName + "/scm-reporting-plr-started"
 )
 
 var ParamsRe = regexp.MustCompile(`{{([^}]{2,})}}`)

pkg/pipelineascode/pipelineascode.go

@@ -257,6 +257,9 @@ func (p *PacRun) startPR(ctx context.Context, match matcher.Match) (*tektonv1.Pi
 		whatPatching = "annotations.state and labels.state"
 		patchAnnotations[keys.State] = kubeinteraction.StateQueued
 		patchLabels[keys.State] = kubeinteraction.StateQueued
+	} else {
+		patchAnnotations[keys.SCMReportingPLRStarted] = "true"
+		whatPatching = fmt.Sprintf("annotation.%s", keys.SCMReportingPLRStarted)
 	}
 
 	if err := p.vcx.CreateStatus(ctx, p.event, status); err != nil {
@@ -278,6 +281,20 @@ func (p *PacRun) startPR(ctx context.Context, match matcher.Match) (*tektonv1.Pi
 			// unneeded SIGSEGV's
 			return pr, fmt.Errorf("cannot patch pipelinerun %s: %w", pr.GetGenerateName(), err)
 		}
+		currentReason := ""
+		if len(pr.Status.GetConditions()) > 0 {
+			currentReason = pr.Status.GetConditions()[0].GetReason()
+		}
+
+		p.logger.Infof("PipelineRun %s/%s patched successfully - Spec.Status: %s, State annotation: '%s', SCMReportingPLRStarted annotation: '%s', Status reason: '%s', Git provider status: '%s', Patched: %s",
+			pr.GetNamespace(),
+			pr.GetName(),
+			pr.Spec.Status,
+			pr.GetAnnotations()[keys.State],
+			pr.GetAnnotations()[keys.SCMReportingPLRStarted],
+			currentReason,
+			status.Status,
+			whatPatching)
 	}
 
 	return pr, nil

pkg/reconciler/reconciler.go

@@ -54,6 +54,23 @@ var (
 func (r *Reconciler) ReconcileKind(ctx context.Context, pr *tektonv1.PipelineRun) pkgreconciler.Event {
 	ctx = info.StoreNS(ctx, system.Namespace())
 	logger := logging.FromContext(ctx).With("namespace", pr.GetNamespace())
+
+	logger.Infof("=== RECONCILE START: PipelineRun %s/%s (resourceVersion: %s) ===", pr.GetNamespace(), pr.GetName(), pr.GetResourceVersion())
+
+	// Get fresh PipelineRun from API server to avoid informer cache staleness
+	freshPR, err := r.run.Clients.Tekton.TektonV1().PipelineRuns(pr.GetNamespace()).Get(ctx, pr.GetName(), metav1.GetOptions{})
+	if err != nil {
+		logger.Errorf("Failed to get fresh PipelineRun %s/%s: %v", pr.GetNamespace(), pr.GetName(), err)
+		return fmt.Errorf("cannot get fresh PipelineRun: %w", err)
+	}
+
+	if freshPR.GetResourceVersion() != pr.GetResourceVersion() {
+		logger.Infof("Using fresh PipelineRun data (cached version %s vs fresh version %s)", pr.GetResourceVersion(), freshPR.GetResourceVersion())
+		pr = freshPR
+	} else {
+		logger.Infof("Cached PipelineRun data is current (resourceVersion: %s)", pr.GetResourceVersion())
+	}
+
 	// if pipelineRun is in completed or failed state then return
 	state, exist := pr.GetAnnotations()[keys.State]
 	if exist && (state == kubeinteraction.StateCompleted || state == kubeinteraction.StateFailed) {
@@ -69,13 +86,27 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, pr *tektonv1.PipelineRun
 	// another controller outside PaC). To maintain consistency between the PipelineRun
 	// status and the Git provider status, we update both the PipelineRun resource and
 	// the corresponding status on the Git provider here.
-	if reason == string(tektonv1.PipelineRunReasonRunning) && state == kubeinteraction.StateQueued {
+	startReported := false
+	scmReportingPLRStarted, exist := pr.GetAnnotations()[keys.SCMReportingPLRStarted]
+	if exist && scmReportingPLRStarted == "true" {
+		startReported = true
+	} else if !exist {
+		logger.Infof("RECONCILE: PipelineRun %s/%s has no %s annotation", pr.GetNamespace(), pr.GetName(), keys.SCMReportingPLRStarted)
+	}
+
+	logger.Infof("RECONCILE: Checking condition - reason='%s' (is Running: %v), state='%s' (is queued: %v) (startReported: %v)",
+		reason, reason == string(tektonv1.PipelineRunReasonRunning), state, state == kubeinteraction.StateQueued, startReported)
+
+	if reason == string(tektonv1.PipelineRunReasonRunning) && !startReported {
+		logger.Infof("RECONCILE: PipelineRun %s/%s condition met (reason=Running, state=queued), calling updatePipelineRunToInProgress", pr.GetNamespace(), pr.GetName())
 		repoName := pr.GetAnnotations()[keys.Repository]
 		repo, err := r.repoLister.Repositories(pr.Namespace).Get(repoName)
 		if err != nil {
 			return fmt.Errorf("failed to get repository CR: %w", err)
 		}
 		return r.updatePipelineRunToInProgress(ctx, logger, repo, pr)
+	} else {
+		logger.Infof("RECONCILE: Condition NOT met for %s/%s - reason='%s', state='%s' (startReported: %v)", pr.GetNamespace(), pr.GetName(), reason, state, startReported)
 	}
 
 	// if its a GitHub App pipelineRun PR then process only if check run id is added otherwise wait
@@ -343,16 +374,24 @@ func (r *Reconciler) updatePipelineRunToInProgress(ctx context.Context, logger *
 }
 
 func (r *Reconciler) updatePipelineRunState(ctx context.Context, logger *zap.SugaredLogger, pr *tektonv1.PipelineRun, state string) (*tektonv1.PipelineRun, error) {
+	currentState := pr.GetAnnotations()[keys.State]
+	logger.Infof("updating pipelineRun %v/%v state from %s to %s", pr.GetNamespace(), pr.GetName(), currentState, state)
+	annotations := map[string]string{
+		keys.State: state,
+	}
+	if state == kubeinteraction.StateStarted {
+		annotations[keys.SCMReportingPLRStarted] = "true"
+	}
+
 	mergePatch := map[string]any{
 		"metadata": map[string]any{
 			"labels": map[string]string{
 				keys.State: state,
 			},
-			"annotations": map[string]string{
-				keys.State: state,
-			},
+			"annotations": annotations,
 		},
 	}
+
 	// if state is started then remove pipelineRun pending status
 	if state == kubeinteraction.StateStarted {
 		mergePatch["spec"] = map[string]any{