fix: ignore duplicate workflow_job completed events for a job already scaled down

controllers/actions.summerwind.net/horizontal_runner_autoscaler_webhook.go

@@ -77,12 +77,47 @@ type HorizontalRunnerAutoscalerGitHubWebhook struct {
 
 	worker     *worker
 	workerInit sync.Once
+
+	scaledDownForMu sync.Mutex
+	scaledDownFor   map[int64]time.Time
 }
 
 func (autoscaler *HorizontalRunnerAutoscalerGitHubWebhook) Reconcile(_ context.Context, request reconcile.Request) (reconcile.Result, error) {
 	return ctrl.Result{}, nil
 }
 
+// scaledDownForDedupWindow bounds how long we remember a job ID we've already scaled
+// down for. GitHub resends "completed" events for jobs from an earlier run when only
+// some jobs are re-run, but a job ID only ever completes once. See #4315.
+const scaledDownForDedupWindow = time.Hour
+
+// markScaledDownFor returns true the first time it sees a job ID, false on any repeat
+// within scaledDownForDedupWindow.
+func (autoscaler *HorizontalRunnerAutoscalerGitHubWebhook) markScaledDownFor(workflowJobID int64) bool {
+	now := time.Now()
+
+	autoscaler.scaledDownForMu.Lock()
+	defer autoscaler.scaledDownForMu.Unlock()
+
+	if autoscaler.scaledDownFor == nil {
+		autoscaler.scaledDownFor = map[int64]time.Time{}
+	}
+
+	for id, seenAt := range autoscaler.scaledDownFor {
+		if now.Sub(seenAt) > scaledDownForDedupWindow {
+			delete(autoscaler.scaledDownFor, id)
+		}
+	}
+
+	if _, seen := autoscaler.scaledDownFor[workflowJobID]; seen {
+		return false
+	}
+
+	autoscaler.scaledDownFor[workflowJobID] = now
+
+	return true
+}
+
 // +kubebuilder:rbac:groups=actions.summerwind.dev,resources=horizontalrunnerautoscalers,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=actions.summerwind.dev,resources=horizontalrunnerautoscalers/finalizers,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=actions.summerwind.dev,resources=horizontalrunnerautoscalers/status,verbs=get;update;patch
@@ -219,6 +254,9 @@ func (autoscaler *HorizontalRunnerAutoscalerGitHubWebhook) Handle(w http.Respons
 				// See example check run completion at https://gist.github.com/nathanklick/268fea6496a4d7b14cecb2999747ef84
 				if e.GetWorkflowJob().GetConclusion() == "success" && e.GetWorkflowJob().RunnerID == nil {
 					log.V(1).Info("Ignoring workflow_job event because it does not relate to a self-hosted runner")
+				} else if jobID := e.GetWorkflowJob().GetID(); !autoscaler.markScaledDownFor(jobID) {
+					// See scaledDownForDedupWindow: this job ID already scaled us down once.
+					log.V(1).Info("Ignoring duplicate workflow_job completed event for a job we already scaled down for", "workflowJob.ID", jobID)
 				} else {
 					// A negative amount is processed in the tryScale func as a scale-down request,
 					// that erases the oldest CapacityReservation with the same amount.

controllers/actions.summerwind.net/horizontal_runner_autoscaler_webhook_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -202,6 +203,123 @@ func TestWebhookWorkflowJob(t *testing.T) {
 	})
 }
 
+// TestWebhookWorkflowJobCompletedDedup reproduces
+// https://github.com/actions/actions-runner-controller/issues/4315: GitHub resends
+// "completed" events for jobs that finished in an earlier run of the same workflow run
+// when only some jobs are re-run via "Re-run failed jobs". A second delivery for a job
+// ID we already scaled down for must be ignored, not scaled down again.
+func TestWebhookWorkflowJobCompletedDedup(t *testing.T) {
+	f, err := os.Open("testdata/org_webhook_workflow_job_payload.json")
+	if err != nil {
+		t.Fatalf("could not open the fixture: %s", err)
+	}
+	defer f.Close()
+
+	var e github.WorkflowJobEvent
+	if err := json.NewDecoder(f).Decode(&e); err != nil {
+		t.Fatalf("invalid json: %s", err)
+	}
+
+	e.Action = github.String("completed")
+	e.WorkflowJob.Status = github.String("completed")
+	e.WorkflowJob.Conclusion = github.String("failure")
+	runnerID := int64(1)
+	e.WorkflowJob.RunnerID = &runnerID
+
+	hra := &actionsv1alpha1.HorizontalRunnerAutoscaler{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-name",
+		},
+		Spec: actionsv1alpha1.HorizontalRunnerAutoscalerSpec{
+			ScaleTargetRef: actionsv1alpha1.ScaleTargetRef{
+				Name: "test-name",
+			},
+			ScaleUpTriggers: []actionsv1alpha1.ScaleUpTrigger{
+				{
+					GitHubEvent: &actionsv1alpha1.GitHubEventScaleUpTriggerSpec{
+						WorkflowJob: &actionsv1alpha1.WorkflowJobSpec{},
+					},
+				},
+			},
+		},
+	}
+
+	rd := &actionsv1alpha1.RunnerDeployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-name",
+		},
+		Spec: actionsv1alpha1.RunnerDeploymentSpec{
+			Template: actionsv1alpha1.RunnerTemplate{
+				Spec: actionsv1alpha1.RunnerSpec{
+					RunnerConfig: actionsv1alpha1.RunnerConfig{
+						Organization: "MYORG",
+						Labels:       []string{"label1"},
+					},
+				},
+			},
+		},
+	}
+
+	hraWebhook := &HorizontalRunnerAutoscalerGitHubWebhook{}
+
+	client := fake.NewClientBuilder().
+		WithScheme(sc).
+		WithRuntimeObjects(hra, rd).
+		WithIndex(&actionsv1alpha1.HorizontalRunnerAutoscaler{}, scaleTargetKey, hraWebhook.indexer).
+		Build()
+
+	logs := installTestLogger(hraWebhook)
+	hraWebhook.Client = client
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", hraWebhook.Handle)
+
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	defer func() {
+		if t.Failed() {
+			t.Logf("diagnostics: %s", logs.String())
+		}
+	}()
+
+	// First delivery: a genuine completion. Must scale down by 1.
+	resp1, err := sendWebhook(server, "workflow_job", &e)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp1.Body.Close()
+
+	body1, err := io.ReadAll(resp1.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if resp1.StatusCode != 200 || string(body1) != "scaled test-name by -1" {
+		t.Fatalf("first delivery: want 200 %q, got %d %q", "scaled test-name by -1", resp1.StatusCode, string(body1))
+	}
+
+	// Second delivery: GitHub resending the same job's completion. Must be ignored.
+	resp2, err := sendWebhook(server, "workflow_job", &e)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp2.Body.Close()
+
+	body2, err := io.ReadAll(resp2.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if resp2.StatusCode != 200 || string(body2) != "" {
+		t.Fatalf("second (duplicate) delivery: want 200 with an empty (ignored) body, got %d %q", resp2.StatusCode, string(body2))
+	}
+
+	if !strings.Contains(logs.String(), "Ignoring duplicate workflow_job completed event") {
+		t.Fatalf("expected a duplicate-detection log line, got: %s", logs.String())
+	}
+}
+
 func TestWebhookWorkflowJobWithSelfHostedLabel(t *testing.T) {
 	setupTest := func() github.WorkflowJobEvent {
 		f, err := os.Open("testdata/org_webhook_workflow_job_with_self_hosted_label_payload.json")