Keep the outdated phase until the runner spec itself changes (#4660)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

apis/actions.github.com/v1alpha1/autoscalinglistener_types.go

@@ -21,8 +21,39 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+// AutoscalingListenerPhase describes whether the listener should be running.
+//
+// It is part of the spec rather than the status because it is a desired state
+// written by the AutoscalingRunnerSet controller, not an observation made by the
+// AutoscalingListener controller about itself.
+type AutoscalingListenerPhase string
+
+const (
+	// AutoscalingListenerPhaseRunning is the default. The listener controller
+	// creates and maintains the listener pod and everything it depends on.
+	AutoscalingListenerPhaseRunning AutoscalingListenerPhase = "Running"
+
+	// AutoscalingListenerPhaseStopped switches the listener off without
+	// deleting it. The listener controller tears the pod and the rest of the
+	// child resources down, so no further jobs are acquired, but the
+	// AutoscalingListener object stays as the record of a scale set that is
+	// meant to come back. Moving the phase to Running is what starts it again.
+	AutoscalingListenerPhaseStopped AutoscalingListenerPhase = "Stopped"
+)
+
+// Stopped reports whether the phase switches the listener off. The zero value
+// means Running, so listeners written before this field existed keep working.
+func (p AutoscalingListenerPhase) Stopped() bool {
+	return p == AutoscalingListenerPhaseStopped
+}
+
 // AutoscalingListenerSpec defines the desired state of AutoscalingListener
 type AutoscalingListenerSpec struct {
+	// Phase controls whether the listener runs. Empty means Running.
+	// +kubebuilder:validation:Enum=Running;Stopped
+	// +optional
+	Phase AutoscalingListenerPhase `json:"phase,omitempty"`
+
 	// +optional
 	GitHubConfigURL string `json:"githubConfigUrl,omitempty"`
 
@@ -94,6 +125,7 @@ type AutoscalingListenerStatus struct{}
 // +kubebuilder:printcolumn:JSONPath=".spec.githubConfigUrl",name=GitHub Configure URL,type=string
 // +kubebuilder:printcolumn:JSONPath=".spec.autoscalingRunnerSetNamespace",name=AutoscalingRunnerSet Namespace,type=string
 // +kubebuilder:printcolumn:JSONPath=".spec.autoscalingRunnerSetName",name=AutoscalingRunnerSet Name,type=string
+// +kubebuilder:printcolumn:JSONPath=".spec.phase",name=Phase,type=string
 
 // AutoscalingListener is the Schema for the autoscalinglisteners API
 type AutoscalingListener struct {

charts/gha-runner-scale-set-controller-experimental/crds/actions.github.com_autoscalinglisteners.yaml

@@ -24,6 +24,9 @@ spec:
     - jsonPath: .spec.autoscalingRunnerSetName
       name: AutoscalingRunnerSet Name
       type: string
+    - jsonPath: .spec.phase
+      name: Phase
+      type: string
     name: v1alpha1
     schema:
       openAPIV3Schema:
@@ -193,6 +196,13 @@ spec:
               minRunners:
                 minimum: 0
                 type: integer
+              phase:
+                description: Phase controls whether the listener runs. Empty means
+                  Running.
+                enum:
+                - Running
+                - Stopped
+                type: string
               proxy:
                 properties:
                   http:

charts/gha-runner-scale-set-controller/crds/actions.github.com_autoscalinglisteners.yaml

@@ -24,6 +24,9 @@ spec:
     - jsonPath: .spec.autoscalingRunnerSetName
       name: AutoscalingRunnerSet Name
       type: string
+    - jsonPath: .spec.phase
+      name: Phase
+      type: string
     name: v1alpha1
     schema:
       openAPIV3Schema:
@@ -193,6 +196,13 @@ spec:
               minRunners:
                 minimum: 0
                 type: integer
+              phase:
+                description: Phase controls whether the listener runs. Empty means
+                  Running.
+                enum:
+                - Running
+                - Stopped
+                type: string
               proxy:
                 properties:
                   http:

config/crd/bases/actions.github.com_autoscalinglisteners.yaml

@@ -24,6 +24,9 @@ spec:
     - jsonPath: .spec.autoscalingRunnerSetName
       name: AutoscalingRunnerSet Name
       type: string
+    - jsonPath: .spec.phase
+      name: Phase
+      type: string
     name: v1alpha1
     schema:
       openAPIV3Schema:
@@ -193,6 +196,13 @@ spec:
               minRunners:
                 minimum: 0
                 type: integer
+              phase:
+                description: Phase controls whether the listener runs. Empty means
+                  Running.
+                enum:
+                - Running
+                - Stopped
+                type: string
               proxy:
                 properties:
                   http:

controllers/actions.github.com/autoscalinglistener_controller.go

@@ -148,6 +148,26 @@ func (r *AutoscalingListenerReconciler) Reconcile(ctx context.Context, req ctrl.
 		return ctrl.Result{}, err
 	}
 
+	// A stopped listener keeps its object, spec and finalizer, but owns nothing:
+	// the AutoscalingRunnerSet parks it this way instead of deleting it so the
+	// scale set stops acquiring jobs without re-registering with the Actions
+	// service when it comes back. Patching the phase back to Running falls
+	// through to the regular reconcile below, which rebuilds the children.
+	if autoscalingListener.Spec.Phase.Stopped() {
+		log.Info("Listener is stopped, cleaning up its resources")
+		requeue, err := r.cleanupResources(ctx, &autoscalingListener, log)
+		if err != nil {
+			log.Error(err, "Failed to clean up the resources of a stopped listener")
+			return ctrl.Result{}, err
+		}
+		if requeue {
+			return ctrl.Result{RequeueAfter: time.Second}, nil
+		}
+
+		log.Info("Listener is stopped and all of its resources are cleaned up")
+		return ctrl.Result{}, nil
+	}
+
 	// Make sure the runner scale set listener service account is created for the listener pod in the controller namespace
 	var serviceAccount corev1.ServiceAccount
 	err := r.Get(
@@ -621,24 +641,29 @@ func (r *AutoscalingListenerReconciler) cleanupResources(ctx context.Context, au
 		return false, fmt.Errorf("failed to get listener config secret: %w", err)
 	}
 
-	if autoscalingListener.Spec.Proxy != nil {
-		logger.Info("Cleaning up the listener proxy secret")
-		proxySecret := new(corev1.Secret)
-		err = r.Get(ctx, types.NamespacedName{Name: proxyListenerSecretName(autoscalingListener), Namespace: autoscalingListener.Namespace}, proxySecret)
-		switch {
-		case err == nil:
-			if proxySecret.DeletionTimestamp.IsZero() {
-				logger.Info("Deleting the listener proxy secret")
-				if err := r.Delete(ctx, proxySecret); err != nil {
-					return false, fmt.Errorf("failed to delete listener proxy secret: %w", err)
-				}
+	// The proxy secret is deleted whatever the current spec says about a proxy.
+	// Its name is derived from the listener rather than from the spec, and the
+	// spec of a stopped listener is updated in place as the AutoscalingRunnerSet
+	// is edited, so asking the spec would let a user who removes the proxy while
+	// the scale set is switched off erase the only signal that the old secret is
+	// there. It holds credentials, and a stopped listener is never deleted, so
+	// nothing else would ever collect it.
+	logger.Info("Cleaning up the listener proxy secret")
+	proxySecret := new(corev1.Secret)
+	err = r.Get(ctx, types.NamespacedName{Name: proxyListenerSecretName(autoscalingListener), Namespace: autoscalingListener.Namespace}, proxySecret)
+	switch {
+	case err == nil:
+		if proxySecret.DeletionTimestamp.IsZero() {
+			logger.Info("Deleting the listener proxy secret")
+			if err := r.Delete(ctx, proxySecret); err != nil {
+				return false, fmt.Errorf("failed to delete listener proxy secret: %w", err)
 			}
-			requeue = true
-		case !kerrors.IsNotFound(err):
-			return false, fmt.Errorf("failed to get listener proxy secret: %w", err)
 		}
-		logger.Info("Listener proxy secret is deleted")
+		requeue = true
+	case !kerrors.IsNotFound(err):
+		return false, fmt.Errorf("failed to get listener proxy secret: %w", err)
 	}
+	logger.Info("Listener proxy secret is deleted")
 
 	listenerRoleBinding := new(rbacv1.RoleBinding)
 	err = r.Get(ctx, types.NamespacedName{Namespace: autoscalingListener.Spec.AutoscalingRunnerSetNamespace, Name: autoscalingListener.Name}, listenerRoleBinding)

controllers/actions.github.com/autoscalinglistener_controller_test.go

@@ -646,6 +646,94 @@ var _ = Describe("Test AutoScalingListener controller", func() {
 			).Should(BeEquivalentTo(oldSecretUID), "Config secret should persist (not be re-created)")
 		})
 	})
+
+	Context("When the listener is stopped", func() {
+		listenerKey := func() client.ObjectKey {
+			return client.ObjectKey{Name: autoscalingListener.Name, Namespace: autoscalingListener.Namespace}
+		}
+
+		// The child resources are the listener's whole footprint: the pod that
+		// acquires jobs, the config secret holding its credentials, and the RBAC
+		// it runs under.
+		//
+		// A pod counts as removed once its deletion has been requested. envtest
+		// runs no kubelet, so nothing confirms the delete and the pod lingers
+		// Terminating for its whole 60s grace period; the controller has already
+		// done everything it can at that point.
+		expectChildResources := func(exist bool) {
+			GinkgoHelper()
+
+			Eventually(
+				func(g Gomega) {
+					pod := new(corev1.Pod)
+					err := k8sClient.Get(ctx, listenerKey(), pod)
+					if exist {
+						g.Expect(err).NotTo(HaveOccurred(), "pod should exist")
+						g.Expect(pod.DeletionTimestamp).To(BeNil(), "pod should not be terminating")
+					} else {
+						g.Expect(kerrors.IsNotFound(err) || (err == nil && pod.DeletionTimestamp != nil)).
+							To(BeTrue(), "pod should be removed while the listener is stopped")
+					}
+
+					for _, resource := range []struct {
+						name   string
+						key    client.ObjectKey
+						object client.Object
+					}{
+						{"config secret", client.ObjectKey{Name: scaleSetListenerConfigName(autoscalingListener), Namespace: autoscalingListener.Namespace}, new(corev1.Secret)},
+						{"service account", listenerKey(), new(corev1.ServiceAccount)},
+						{"role", client.ObjectKey{Name: autoscalingListener.Name, Namespace: autoscalingListener.Spec.AutoscalingRunnerSetNamespace}, new(rbacv1.Role)},
+						{"role binding", client.ObjectKey{Name: autoscalingListener.Name, Namespace: autoscalingListener.Spec.AutoscalingRunnerSetNamespace}, new(rbacv1.RoleBinding)},
+					} {
+						err := k8sClient.Get(ctx, resource.key, resource.object)
+						if exist {
+							g.Expect(err).NotTo(HaveOccurred(), "%s should exist", resource.name)
+							continue
+						}
+						g.Expect(kerrors.IsNotFound(err)).To(BeTrue(), "%s should be removed while the listener is stopped", resource.name)
+					}
+				},
+				autoscalingListenerTestTimeout,
+				autoscalingListenerTestInterval,
+			).Should(Succeed())
+		}
+
+		patchPhase := func(phase v1alpha1.AutoscalingListenerPhase) {
+			GinkgoHelper()
+
+			listener := new(v1alpha1.AutoscalingListener)
+			Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed())
+			original := listener.DeepCopy()
+			listener.Spec.Phase = phase
+			Expect(k8sClient.Patch(ctx, listener, client.MergeFrom(original))).To(Succeed(), "failed to patch the listener phase")
+		}
+
+		It("removes the child resources but keeps the listener, and rebuilds them when started again", func() {
+			expectChildResources(true)
+
+			patchPhase(v1alpha1.AutoscalingListenerPhaseStopped)
+
+			expectChildResources(false)
+
+			// The listener itself is the record of a scale set that is meant to
+			// come back, so it survives with its finalizer and spec intact.
+			Consistently(
+				func(g Gomega) {
+					listener := new(v1alpha1.AutoscalingListener)
+					g.Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed())
+					g.Expect(listener.DeletionTimestamp).To(BeNil(), "a stopped listener must not be deleted")
+					g.Expect(listener.Finalizers).To(ContainElement(autoscalingListenerFinalizerName))
+					g.Expect(listener.Spec.Phase).To(Equal(v1alpha1.AutoscalingListenerPhaseStopped))
+				},
+				2*time.Second,
+				autoscalingListenerTestInterval,
+			).Should(Succeed())
+
+			patchPhase(v1alpha1.AutoscalingListenerPhaseRunning)
+
+			expectChildResources(true)
+		})
+	})
 })
 
 var _ = Describe("Test AutoScalingListener customization", func() {

controllers/actions.github.com/autoscalinglistener_proxy_cleanup_test.go

@@ -0,0 +1,79 @@
+package actionsgithubcom
+
+import (
+	"context"
+	"testing"
+
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
+)
+
+// The proxy secret holds credentials and is owned by the listener, so deleting
+// the listener collects it. A stopped listener is not deleted, which makes
+// cleanupResources the only thing that will ever remove it.
+//
+// Its name is derived from the listener, not from the spec, so cleanup must not
+// ask the spec whether it exists. The spec of a parked listener is updated in
+// place as the AutoscalingRunnerSet is edited, so a user who removes the proxy
+// while the scale set is switched off would otherwise erase the only signal that
+// the old secret is there, and leave the credentials behind indefinitely.
+func TestCleanupResourcesDeletesTheProxySecretWithoutAskingTheSpec(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, clientgoscheme.AddToScheme(scheme))
+	require.NoError(t, v1alpha1.AddToScheme(scheme))
+
+	const namespace = "arc-system"
+
+	listener := &v1alpha1.AutoscalingListener{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-listener", Namespace: namespace},
+		Spec: v1alpha1.AutoscalingListenerSpec{
+			AutoscalingRunnerSetNamespace: "arc-runners",
+			AutoscalingRunnerSetName:      "test-ars",
+			Phase:                         v1alpha1.AutoscalingListenerPhaseStopped,
+			// The proxy has been removed from the scale set since the secret was
+			// created, and the parked listener has already taken that edit.
+			Proxy: nil,
+		},
+	}
+	proxySecret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      proxyListenerSecretName(listener),
+			Namespace: namespace,
+		},
+		Data: map[string][]byte{"password": []byte("secret")},
+	}
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(listener, proxySecret).
+		Build()
+
+	reconciler := &AutoscalingListenerReconciler{
+		Client: fakeClient,
+		Scheme: scheme,
+		Log:    logr.Discard(),
+	}
+
+	_, err := reconciler.cleanupResources(context.Background(), listener, logr.Discard())
+	require.NoError(t, err)
+
+	err = fakeClient.Get(
+		context.Background(),
+		types.NamespacedName{Name: proxySecret.Name, Namespace: namespace},
+		new(corev1.Secret),
+	)
+	require.True(
+		t,
+		kerrors.IsNotFound(err),
+		"the proxy secret must be removed with the rest of the listener's resources, whatever the current spec says about a proxy",
+	)
+}

controllers/actions.github.com/autoscalinglistener_restart_convergence_test.go

@@ -0,0 +1,37 @@
+package actionsgithubcom
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+)
+
+// A stop the listener controller never observed - because the scale set
+// recovered before it reconciled, and the queue coalesced the two phase writes -
+// leaves the listener pod in place. Nothing re-runs the cleanup that was skipped,
+// so the restart has to converge the pod on its own: what the scale set comes
+// back with must be the pod its current spec describes, not the one it was
+// parked with. Reconciliation is level-triggered, so the comparison below is
+// what makes the skipped stop harmless rather than a stale listener.
+func TestAStoppedListenerThatKeptItsPodStillConvergesOnRestart(t *testing.T) {
+	current := &corev1.Pod{Spec: corev1.PodSpec{Containers: []corev1.Container{{
+		Name:  "listener",
+		Image: "listener:image",
+		Env:   []corev1.EnvVar{{Name: "MIN_RUNNERS", Value: "1"}},
+	}}}}
+
+	desired := current.DeepCopy()
+	desired.Spec.Containers[0].Env[0].Value = "2"
+
+	require.True(
+		t,
+		listenerPodSpecRequiresRecreation(current, desired),
+		"a pod left over from before the stop must be rebuilt when the spec moved on while parked",
+	)
+	require.False(
+		t,
+		listenerPodSpecRequiresRecreation(current, current.DeepCopy()),
+		"a pod that already matches must be kept, so a skipped stop is not a reason to churn it",
+	)
+}

controllers/actions.github.com/autoscalingrunnerset_controller.go

@@ -145,6 +145,45 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 		return ctrl.Result{}, nil
 	}
 
+	// The outdated phase is sticky. It means the runners rejected the runner
+	// spec they were given, so the only edit worth retrying is one that changes
+	// what the next runner would be handed: the runner spec, or the metadata
+	// stamped onto it. Every other edit - replica bounds, runner group, scale
+	// set name - bumps metadata.generation without changing anything the runners
+	// objected to, and acting on it would switch the listener back on to acquire
+	// jobs for runners that will reject the spec exactly as before.
+	//
+	// This is therefore checked before the generation comparison below, which
+	// would otherwise move the phase to pending for any spec edit at all and
+	// undo the teardown.
+	if autoscalingRunnerSet.Status.Phase == v1alpha1.AutoscalingRunnerSetPhaseOutdated {
+		corrected, err := r.outdatedRunnerSpecCorrected(ctx, &autoscalingRunnerSet, log)
+		if err != nil {
+			log.Error(err, "Failed to compare the outdated runner spec with the desired one")
+			return ctrl.Result{}, err
+		}
+
+		if !corrected {
+			return r.reconcileOutdated(ctx, &autoscalingRunnerSet, log)
+		}
+
+		// The runner spec changed, so the scale set may run again. Move to
+		// pending and let the reconcile below publish the new spec: that patch
+		// also advances the actionable revision, which is what tells the
+		// EphemeralRunnerSet to stop judging itself by the runners that failed.
+		log.Info("Runner spec of an outdated autoscaling runner set changed. Recovering from the outdated phase")
+		if err := r.updateStatus(
+			ctx,
+			&autoscalingRunnerSet,
+			v1alpha1.AutoscalingRunnerSetPhasePending,
+			autoscalingRunnerSet.Status.ObservedGeneration,
+			log,
+		); err != nil {
+			log.Error(err, "Failed to update autoscaling runner set status with pending phase")
+			return ctrl.Result{}, err
+		}
+	}
+
 	// The spec changed since we last observed it, so move back to the pending
 	// phase. The observed generation is deliberately left at its old value here:
 	// it only catches up at the end of a successful reconcile, so a reconcile
@@ -163,10 +202,6 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 		}
 	}
 
-	if autoscalingRunnerSet.Status.Phase == v1alpha1.AutoscalingRunnerSetPhaseOutdated {
-		return r.reconcileOutdated(ctx, &autoscalingRunnerSet, log)
-	}
-
 	if shouldCreateScaleSet(&autoscalingRunnerSet) {
 		log.Info("Creating runner scale set")
 		return r.createRunnerScaleSet(ctx, &autoscalingRunnerSet, log)
@@ -203,14 +238,14 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 		log.Error(err, "Failed to get ephemeral runner")
 		return ctrl.Result{}, err
 	case ephemeralRunnerSetOutdatedForAppliedRevision(&ephemeralRunnerSet) &&
-		!ephemeralRunnerSetNeedsOutdatedRecovery(&ephemeralRunnerSet, &autoscalingRunnerSet):
+		!r.runnerSpecChanged(&autoscalingRunnerSet, &ephemeralRunnerSet, log):
 		// The runners rejected the spec they were given, so the scale set has to
 		// stop acquiring jobs it cannot run. Record that in the phase first: it is
 		// what keeps the listener switched off across reconciles, and what stops
 		// the branches below from rebuilding it. This also covers Pending during a
-		// metadata-only listener rebuild; only an unobserved spec generation is a
-		// recovery signal. The observed generation is carried over unchanged, so a
-		// spec update still registers as new work.
+		// metadata-only listener rebuild, which leaves the runner spec untouched
+		// and so is not a recovery signal. The observed generation is carried over
+		// unchanged, so a spec update still registers as new work.
 		log.Info("Ephemeral runner set is outdated. Moving the autoscaling runner set to the outdated phase")
 		if err := r.updateStatus(
 			ctx,
@@ -232,18 +267,15 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 			return ctrl.Result{}, nil
 		}
 
+		// Recovering from the outdated phase has to advance the revision even
+		// when only the runner metadata changed. The revision is what tells the
+		// EphemeralRunnerSet to stop judging itself by the runners that failed,
+		// clearing its outdated phase and allowing it to scale up again; without
+		// it the metadata patch below would land and the set would be pushed
+		// straight back to outdated.
 		recoveringFromOutdated := ephemeralRunnerSetOutdatedForAppliedRevision(&ephemeralRunnerSet) &&
-			ephemeralRunnerSetNeedsOutdatedRecovery(&ephemeralRunnerSet, &autoscalingRunnerSet)
+			ephemeralRunnerSetDesiredSpecChanged(&ephemeralRunnerSet, desired)
 		if ephemeralRunnerSetActionableSpecChanged(&ephemeralRunnerSet, desired) || recoveringFromOutdated {
-			// A real AutoscalingRunnerSet spec update leaves its observed generation
-			// behind until reconciliation succeeds. Require that signal before
-			// recovering an outdated set: Pending can also mean a metadata-only
-			// listener rebuild, which must not retry the same rejected runner spec.
-			//
-			// The revision has to advance even when the runner spec itself is
-			// unchanged. It tells the EphemeralRunnerSet to stop judging itself by
-			// the runners that failed, clearing its outdated phase and allowing it
-			// to scale up again.
 			original := ephemeralRunnerSet.DeepCopy()
 			ephemeralRunnerSet.Spec.EphemeralRunnerMetadata = desired.Spec.EphemeralRunnerMetadata
 			ephemeralRunnerSet.Spec.EphemeralRunnerSpec = desired.Spec.EphemeralRunnerSpec
@@ -287,6 +319,18 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 		}
 	}
 
+	// Renamed listeners are dropped before the lookup below, which is by the
+	// current name and so cannot see them. Leaving one in place would let the
+	// scale set run two listeners at once: the replacement created here, and the
+	// one still holding a pod under the name the scale set used to derive.
+	switch deleted, err := r.deleteRenamedListeners(ctx, &autoscalingRunnerSet, log); {
+	case err != nil:
+		log.Error(err, "Failed to delete the listeners left behind under a previous name")
+		return ctrl.Result{}, err
+	case deleted:
+		return ctrl.Result{}, nil
+	}
+
 	var listener v1alpha1.AutoscalingListener
 	err = r.Get(
 		ctx,
@@ -319,14 +363,22 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 			&ephemeralRunnerSet,
 			r.ControllerNamespace,
 			r.DefaultRunnerScaleSetListenerImage,
-			nil, // TODO: remove
+			r.listenerImagePullSecrets(),
 		)
 		if err != nil {
 			log.Error(err, "Failed to generate AutoscalingListener spec")
 			return ctrl.Result{}, nil
 		}
 
-		if !cmp.Equal(listener.Spec, desired.Spec) ||
+		// The drift check has to come before the phase is started again. While a
+		// scale set is parked, reconcileOutdated returns early and no listener
+		// spec drift is propagated, yet edits outside the runner spec are still
+		// allowed in that window. Starting first would rebuild the pod and its
+		// children from the spec the listener was parked with, and let it
+		// acquire jobs under it until the next reconcile noticed. Deleting
+		// instead re-creates the listener with the phase unset, which means
+		// running, so it comes back correct in one step.
+		if listenerSpecChanged(&listener, desired) ||
 			!cmp.Equal(listener.Labels, desired.Labels) ||
 			!cmp.Equal(listener.Annotations, desired.Annotations) {
 			// The listener is about to be torn down and rebuilt, which is what
@@ -356,6 +408,23 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 			log.Info("Deleted AutoscalingListener, will re-create on next reconcile")
 			return ctrl.Result{}, nil
 		}
+
+		// The spec already matches, so a listener that was switched off may run
+		// again as it stands. Start it by moving the phase rather than by
+		// replacing the object: the listener controller rebuilds the pod and the
+		// rest of the child resources from it.
+		if listener.Spec.Phase.Stopped() {
+			log.Info("Starting the stopped listener")
+			original := listener.DeepCopy()
+			listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseRunning
+			if err := r.Patch(ctx, &listener, client.MergeFrom(original)); err != nil {
+				log.Error(err, "Failed to start the stopped listener")
+				return ctrl.Result{}, err
+			}
+
+			log.Info("Started the stopped listener")
+			return ctrl.Result{}, nil
+		}
 	}
 
 	log.Info("Autoscaling runner set is up to date and ready")
@@ -373,33 +442,362 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
 	return ctrl.Result{}, nil
 }
 
-// reconcileOutdated holds a scale set whose runners rejected the runner spec.
+// outdatedRunnerSpecCorrected reports whether the runner spec that the runners
+// rejected has since been changed, which is the only thing that takes a scale
+// set out of the outdated phase.
 //
-// The listener is removed so no new jobs are acquired, and the EphemeralRunnerSet
-// is pinned to zero replicas so it releases every runner that is not currently
-// executing a job. The set itself is deliberately kept: the user has not asked
-// for the scale set to go away, and deleting it would make the controller
-// immediately rebuild it from the same rejected spec, in a loop. Keeping it also
-// preserves the revision bookkeeping that decides when the scale set may run
+// Keying recovery on metadata.generation instead would recover on any spec edit
+// at all, including ones that leave the runner spec untouched, and hand the
+// listener back to a scale set whose runners will reject the very same spec
 // again.
 //
-// This state is left only when the AutoscalingRunnerSet spec is updated, which
-// moves the phase back to pending and lets the next reconcile publish the new
-// spec to the set.
-func (r *AutoscalingRunnerSetReconciler) reconcileOutdated(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, log logr.Logger) (ctrl.Result, error) {
-	log.Info("Autoscaling runner set is in outdated phase, removing the listener")
-	done, err := r.cleanupListener(ctx, autoscalingRunnerSet, log)
-	if err != nil {
-		log.Error(err, "Failed to clean up listener")
-		return ctrl.Result{}, err
+// A missing EphemeralRunnerSet counts as corrected. There is nothing left to
+// compare against, and the set is only absent because something outside the
+// controller removed it, so the reconcile is allowed to rebuild it from the
+// current spec rather than sitting in a phase it could never leave.
+func (r *AutoscalingRunnerSetReconciler) outdatedRunnerSpecCorrected(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, log logr.Logger) (bool, error) {
+	var ephemeralRunnerSet v1alpha1.EphemeralRunnerSet
+	err := r.Get(
+		ctx,
+		types.NamespacedName{
+			Namespace: autoscalingRunnerSet.Namespace,
+			Name:      autoscalingRunnerSet.Name,
+		},
+		&ephemeralRunnerSet,
+	)
+	switch {
+	case kerrors.IsNotFound(err):
+		return true, nil
+	case err != nil:
+		return false, err
 	}
-	if !done {
-		log.Info("Waiting for listener to be cleaned up for the outdated runner set")
-		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+
+	return r.runnerSpecChanged(autoscalingRunnerSet, &ephemeralRunnerSet, log), nil
+}
+
+// runnerSpecChanged compares the part of the EphemeralRunnerSet spec the
+// AutoscalingRunnerSet owns - the runner spec and the metadata stamped onto the
+// runners - with what the set is currently running.
+//
+// A spec that cannot be built counts as unchanged. The comparison is used to
+// decide whether a rejected runner spec may be retried, and an unusable desired
+// spec is no evidence that it was corrected.
+func (r *AutoscalingRunnerSetReconciler) runnerSpecChanged(autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet, log logr.Logger) bool {
+	desired, err := r.newEphemeralRunnerSet(autoscalingRunnerSet)
+	if err != nil {
+		log.Error(err, "Failed to generate ephemeral runner set spec to compare against the rejected runner spec")
+		return false
+	}
+
+	return ephemeralRunnerSetDesiredSpecChanged(ephemeralRunnerSet, desired)
+}
+
+// listenerSpecChanged reports whether the live listener spec differs from the
+// desired one in a way that requires replacing the listener.
+//
+// Phase is excluded. It is the one field the AutoscalingRunnerSet controller
+// writes onto a live listener rather than deriving from its own spec, so
+// comparing it would make a stopped listener look like drift and delete the very
+// object the stop is meant to preserve. Starting and stopping is handled by
+// patching the phase instead.
+func listenerSpecChanged(current, desired *v1alpha1.AutoscalingListener) bool {
+	if current == nil || desired == nil {
+		return current != desired
+	}
+
+	currentSpec := current.Spec
+	desiredSpec := desired.Spec
+	currentSpec.Phase = ""
+	desiredSpec.Phase = ""
+
+	return !cmp.Equal(currentSpec, desiredSpec)
+}
+
+// stopListener switches the listener off without deleting it.
+//
+// The child resources go either way: the listener controller tears down the pod,
+// the config secret and the RBAC once the phase is Stopped, exactly as deleting
+// the listener would have. What survives is the AutoscalingListener object
+// itself, with its spec and finalizer, so a parked scale set stays visible as
+// Phase: Stopped rather than as a listener that silently does not exist, and the
+// custom resource is not churned every time a scale set is parked and recovered.
+// Listeners are found by their back-reference to the scale set rather than by
+// the name derived from it. The derived name is a hash over the runner group and
+// the config URL, so editing either renames the listener the controller looks
+// for - and a parked scale set accepts exactly those edits. Looking the listener
+// up by name would miss the one still running under the previous name and leave
+// it acquiring jobs for a scale set that is supposed to be switched off.
+func (r *AutoscalingRunnerSetReconciler) stopListener(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, log logr.Logger) error {
+	listeners, err := r.listenersForAutoscalingRunnerSet(ctx, autoscalingRunnerSet)
+	if err != nil {
+		return err
+	}
+
+	// Nothing to switch off. A scale set that is switched off never creates a
+	// listener, so this is the normal state after a restart.
+	for i := range listeners {
+		listener := &listeners[i]
+		if !listener.DeletionTimestamp.IsZero() || listener.Spec.Phase.Stopped() {
+			continue
+		}
+
+		log.Info("Stopping the listener so no further jobs are acquired", "listener", listener.Name)
+		original := listener.DeepCopy()
+		listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+		if err := r.Patch(ctx, listener, client.MergeFrom(original)); err != nil {
+			return err
+		}
+
+		log.Info("Stopped the listener", "listener", listener.Name)
+	}
+
+	return nil
+}
+
+// listenersForAutoscalingRunnerSet returns every listener that names this scale
+// set as its own, which is the same back-reference the controller's watch uses
+// to map a listener back to the set that owns it.
+func (r *AutoscalingRunnerSetReconciler) listenersForAutoscalingRunnerSet(
+	ctx context.Context,
+	autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet,
+) ([]v1alpha1.AutoscalingListener, error) {
+	var list v1alpha1.AutoscalingListenerList
+	if err := r.List(
+		ctx,
+		&list,
+		client.InNamespace(r.ControllerNamespace),
+		client.MatchingFields{
+			autoscalingRunnerSetOwnerKey: autoscalingRunnerSetOwnerIndexValue(
+				autoscalingRunnerSet.Namespace,
+				autoscalingRunnerSet.Name,
+			),
+		},
+	); err != nil {
+		return nil, fmt.Errorf("failed to list listeners: %w", err)
+	}
+
+	return list.Items, nil
+}
+
+// deleteRenamedListeners removes the listeners a scale set owns that no longer
+// answer to the name derived from it.
+//
+// That name is a hash over the runner group and the config URL, so editing
+// either renames the listener the rest of the lifecycle looks for, and the
+// listener created under the previous name becomes unreachable: nothing gets,
+// updates or deletes it again while the scale set lives. It is not inert,
+// though - it keeps its pod, and so keeps acquiring jobs alongside whatever
+// replaces it. Deleting it here is what makes a rename a replacement rather than
+// an addition.
+func (r *AutoscalingRunnerSetReconciler) deleteRenamedListeners(
+	ctx context.Context,
+	autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet,
+	log logr.Logger,
+) (deleted bool, err error) {
+	listeners, err := r.listenersForAutoscalingRunnerSet(ctx, autoscalingRunnerSet)
+	if err != nil {
+		return false, err
+	}
+
+	currentName := scaleSetListenerName(autoscalingRunnerSet)
+	for i := range listeners {
+		listener := &listeners[i]
+		if listener.Name == currentName {
+			continue
+		}
+
+		deleted = true
+		if !listener.DeletionTimestamp.IsZero() {
+			continue
+		}
+
+		log.Info("Deleting a listener left behind under a previous name", "listener", listener.Name)
+		if err := r.Delete(ctx, listener); err != nil && !kerrors.IsNotFound(err) {
+			return true, fmt.Errorf("failed to delete renamed listener %q: %w", listener.Name, err)
+		}
+	}
+
+	return deleted, nil
+}
+
+// propagateToStoppedListener brings a switched-off listener's spec up to date.
+//
+// Switched off is not the same as frozen. A parked scale set still accepts edits
+// that are not a recovery signal - replica bounds, labels, annotations - and
+// those belong to the listener even though it is not running. Landing them now
+// means the listener that eventually starts is built from the spec as it stands
+// then, rather than from the spec it was parked with.
+//
+// The spec is patched in place rather than being replaced the way drift is
+// handled on the running path. Replacing exists to rebuild the pod; a stopped
+// listener has no pod, and re-creating the object would bring it back with the
+// phase unset, which means running.
+//
+// Listeners left behind under a previous name are deleted rather than updated.
+// Every path that could start one looks it up by the name derived now, so a
+// stale-named listener can never run again: updating it would only keep a
+// permanent orphan in step with a spec it will never use. Deleting it also means
+// recovery builds the listener fresh, from the name and spec as they stand then.
+func (r *AutoscalingRunnerSetReconciler) propagateToStoppedListener(
+	ctx context.Context,
+	autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet,
+	ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet,
+	log logr.Logger,
+) error {
+	listeners, err := r.listenersForAutoscalingRunnerSet(ctx, autoscalingRunnerSet)
+	if err != nil {
+		return err
+	}
+
+	currentName := scaleSetListenerName(autoscalingRunnerSet)
+	var current *v1alpha1.AutoscalingListener
+	renamed := false
+	for i := range listeners {
+		if listeners[i].Name == currentName {
+			current = &listeners[i]
+			continue
+		}
+		renamed = true
+	}
+
+	switch {
+	case current == nil && !renamed:
+		// The scale set has no listener at all, which is not something an edit
+		// caused: a rejected runner spec is never rebuilt into a listener, so
+		// there is nothing here to bring up to date.
+		return nil
+	case current == nil:
+		// The listener the scale set does have answers to a previous name, so an
+		// edit renamed it. Create the replacement stopped rather than waiting
+		// for recovery to do it: a parked scale set is meant to be visible as a
+		// listener that exists and is switched off, and that should not stop
+		// being true because the user edited the runner group.
+		//
+		// The replacement is created before the listener under the previous name
+		// is deleted, which is the opposite order to the running path. Here the
+		// risk being avoided is a parked scale set that momentarily has no
+		// listener at all; overlapping briefly costs nothing, because both
+		// objects are stopped and a stopped listener has no pod. On the running
+		// path the order is reversed for the same reason read the other way:
+		// overlapping there would mean two listeners acquiring jobs at once.
+		if err := r.createStoppedListener(ctx, autoscalingRunnerSet, ephemeralRunnerSet, log); err != nil {
+			return err
+		}
+
+		_, err := r.deleteRenamedListeners(ctx, autoscalingRunnerSet, log)
+		return err
+	}
+
+	listener := *current
+	if !listener.DeletionTimestamp.IsZero() {
+		_, err := r.deleteRenamedListeners(ctx, autoscalingRunnerSet, log)
+		return err
+	}
+
+	desired, err := r.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		r.ControllerNamespace,
+		r.DefaultRunnerScaleSetListenerImage,
+		r.listenerImagePullSecrets(),
+	)
+	if err != nil {
+		return err
+	}
+
+	desiredLabels := r.filterAndMergeLabels(listener.Labels, desired.Labels)
+	desiredAnnotations := r.mergeAnnotations(listener.Annotations, desired.Annotations)
+	if !listenerSpecChanged(&listener, desired) &&
+		maps.Equal(listener.Labels, desiredLabels) &&
+		maps.Equal(listener.Annotations, desiredAnnotations) {
+		_, err := r.deleteRenamedListeners(ctx, autoscalingRunnerSet, log)
+		return err
+	}
+
+	log.Info("Updating the stopped listener to match the desired spec")
+	original := listener.DeepCopy()
+	listener.Spec = desired.Spec
+	// The phase is forced rather than carried over from the live object. This
+	// helper only runs on the parked path, immediately after stopListener has
+	// patched the phase, and the read above goes through the cache: it can still
+	// return the pre-patch object. Preserving what it reported would write
+	// Running back onto a listener this same reconcile has just switched off.
+	// The desired listener says nothing about the phase either, since it is
+	// written onto the live object rather than derived from the
+	// AutoscalingRunnerSet.
+	listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+	listener.Labels = desiredLabels
+	listener.Annotations = desiredAnnotations
+	if err := r.Patch(ctx, &listener, client.MergeFrom(original)); err != nil {
+		return err
+	}
+
+	log.Info("Updated the stopped listener")
+
+	_, err = r.deleteRenamedListeners(ctx, autoscalingRunnerSet, log)
+	return err
+}
+
+// createStoppedListener creates the listener a parked scale set should have, in
+// the stopped phase, so it never runs a pod.
+func (r *AutoscalingRunnerSetReconciler) createStoppedListener(
+	ctx context.Context,
+	autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet,
+	ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet,
+	log logr.Logger,
+) error {
+	desired, err := r.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		r.ControllerNamespace,
+		r.DefaultRunnerScaleSetListenerImage,
+		r.listenerImagePullSecrets(),
+	)
+	if err != nil {
+		return err
+	}
+
+	// Copied before the phase is stamped on. newAutoscalingListener serves a
+	// shared pointer out of the resource cache, so mutating what it returns
+	// switches off the desired listener every later caller derives, not just
+	// this one. Create would write the resulting object's identity back into the
+	// same shared entry for the same reason.
+	desired = desired.DeepCopy()
+	desired.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+	log.Info("Creating the listener of a parked scale set in the stopped phase", "listener", desired.Name)
+	if err := r.Create(ctx, desired); err != nil && !kerrors.IsAlreadyExists(err) {
+		return fmt.Errorf("failed to create the stopped listener: %w", err)
+	}
+
+	return nil
+}
+
+// reconcileOutdated holds a scale set whose runners rejected the runner spec.
+//
+// The listener is stopped so no new jobs are acquired, and the EphemeralRunnerSet
+// is pinned to zero replicas so it releases every runner that is not currently
+// executing a job. Neither object is deleted: the user has not asked for the
+// scale set to go away, and rebuilding it would only publish the same rejected
+// spec again, in a loop. Keeping them also preserves the revision bookkeeping
+// that decides when the scale set may run again.
+//
+// This state is left only when the runner spec the runners rejected is changed,
+// which moves the phase back to pending and lets the next reconcile publish the
+// new spec to the set and start the listener again.
+//
+// Switched off is not frozen, though. Edits that are not a recovery signal are
+// still propagated to both objects while they are parked, so the scale set that
+// eventually recovers is the one the user has been editing rather than the one
+// it was parked as.
+func (r *AutoscalingRunnerSetReconciler) reconcileOutdated(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, log logr.Logger) (ctrl.Result, error) {
+	log.Info("Autoscaling runner set is in outdated phase, stopping the listener")
+	if err := r.stopListener(ctx, autoscalingRunnerSet, log); err != nil {
+		log.Error(err, "Failed to stop the listener for the outdated runner set")
+		return ctrl.Result{}, err
 	}
 
 	var ephemeralRunnerSet v1alpha1.EphemeralRunnerSet
-	err = r.Get(
+	err := r.Get(
 		ctx,
 		types.NamespacedName{
 			Namespace: autoscalingRunnerSet.Namespace,
@@ -425,13 +823,38 @@ func (r *AutoscalingRunnerSetReconciler) reconcileOutdated(ctx context.Context,
 			return ctrl.Result{}, nil
 		}
 
-		if ephemeralRunnerSet.Spec.Replicas == 0 && ephemeralRunnerSet.Spec.PatchID == 0 {
+		if err := r.propagateToStoppedListener(ctx, autoscalingRunnerSet, &ephemeralRunnerSet, log); err != nil {
+			log.Error(err, "Failed to update the stopped listener for the outdated runner set")
+			return ctrl.Result{}, err
+		}
+
+		// Labels and annotations are all that can differ here. The runner spec
+		// and the runner metadata are recovery signals, so if either had changed
+		// this reconcile would have taken the recovery path instead of this one,
+		// and publishing them from here would hand the runners a new spec without
+		// the revision bump that tells the set to stop judging itself by the
+		// runners that failed.
+		desired, err := r.newEphemeralRunnerSet(autoscalingRunnerSet)
+		if err != nil {
+			log.Error(err, "Failed to generate ephemeral runner set spec for the outdated runner set")
+			return ctrl.Result{}, err
+		}
+
+		desiredLabels := r.filterAndMergeLabels(ephemeralRunnerSet.Labels, desired.Labels)
+		desiredAnnotations := r.mergeAnnotations(ephemeralRunnerSet.Annotations, desired.Annotations)
+
+		pinned := ephemeralRunnerSet.Spec.Replicas == 0 && ephemeralRunnerSet.Spec.PatchID == 0
+		if pinned &&
+			maps.Equal(ephemeralRunnerSet.Labels, desiredLabels) &&
+			maps.Equal(ephemeralRunnerSet.Annotations, desiredAnnotations) {
 			return ctrl.Result{}, nil
 		}
 
 		original := ephemeralRunnerSet.DeepCopy()
 		ephemeralRunnerSet.Spec.Replicas = 0
 		ephemeralRunnerSet.Spec.PatchID = 0
+		ephemeralRunnerSet.Labels = desiredLabels
+		ephemeralRunnerSet.Annotations = desiredAnnotations
 		if err := r.Patch(ctx, &ephemeralRunnerSet, client.MergeFrom(original)); err != nil {
 			log.Error(err, "Failed to patch ephemeral runner set with 0 replicas and reset patch ID for the outdated runner set")
 			return ctrl.Result{}, err
@@ -501,32 +924,36 @@ func (r *AutoscalingRunnerSetReconciler) updateStatus(
 	return nil
 }
 
+// Every listener the scale set owns is waited on, not just the one answering to
+// the name derived from it now. This gates the removal of the scale set's
+// finalizer, and a listener left behind under a previous name still has a pod:
+// reporting it gone would tear the scale set down while it is still acquiring
+// jobs.
 func (r *AutoscalingRunnerSetReconciler) cleanupListener(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, logger logr.Logger) (done bool, err error) {
 	logger.Info("Cleaning up the listener")
-	var listener v1alpha1.AutoscalingListener
-	err = r.Get(
-		ctx,
-		client.ObjectKey{
-			Namespace: r.ControllerNamespace,
-			Name:      scaleSetListenerName(autoscalingRunnerSet),
-		},
-		&listener,
-	)
-	switch {
-	case err == nil:
-		if listener.DeletionTimestamp.IsZero() {
-			logger.Info("Deleting the listener")
-			if err := r.Delete(ctx, &listener); err != nil {
-				return false, fmt.Errorf("failed to delete listener: %w", err)
-			}
-		}
-		return false, nil
-	case !kerrors.IsNotFound(err):
-		return false, fmt.Errorf("failed to get listener: %w", err)
+	listeners, err := r.listenersForAutoscalingRunnerSet(ctx, autoscalingRunnerSet)
+	if err != nil {
+		return false, err
 	}
 
-	logger.Info("Listener is deleted")
-	return true, nil
+	if len(listeners) == 0 {
+		logger.Info("Listener is deleted")
+		return true, nil
+	}
+
+	for i := range listeners {
+		listener := &listeners[i]
+		if !listener.DeletionTimestamp.IsZero() {
+			continue
+		}
+
+		logger.Info("Deleting the listener", "listener", listener.Name)
+		if err := r.Delete(ctx, listener); err != nil && !kerrors.IsNotFound(err) {
+			return false, fmt.Errorf("failed to delete listener %q: %w", listener.Name, err)
+		}
+	}
+
+	return false, nil
 }
 
 func (r *AutoscalingRunnerSetReconciler) cleanupEphemeralRunnerSet(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, logger logr.Logger) (done bool, err error) {
@@ -834,7 +1261,13 @@ func (r *AutoscalingRunnerSetReconciler) createEphemeralRunnerSet(ctx context.Co
 	return ctrl.Result{}, nil
 }
 
-func (r *AutoscalingRunnerSetReconciler) createAutoScalingListenerForRunnerSet(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet, log logr.Logger) (ctrl.Result, error) {
+// listenerImagePullSecrets returns the credentials the listener image is pulled
+// with. They come from controller configuration rather than from the
+// AutoscalingRunnerSet, so every caller that derives a desired listener has to
+// supply them: a caller that leaves them out does not describe a listener
+// without credentials, it describes a listener whose credentials it forgot, and
+// anything comparing against it reads the difference as drift.
+func (r *AutoscalingRunnerSetReconciler) listenerImagePullSecrets() []corev1.LocalObjectReference {
 	var imagePullSecrets []corev1.LocalObjectReference
 	for _, imagePullSecret := range r.DefaultRunnerScaleSetListenerImagePullSecrets {
 		imagePullSecrets = append(imagePullSecrets, corev1.LocalObjectReference{
@@ -842,6 +1275,12 @@ func (r *AutoscalingRunnerSetReconciler) createAutoScalingListenerForRunnerSet(c
 		})
 	}
 
+	return imagePullSecrets
+}
+
+func (r *AutoscalingRunnerSetReconciler) createAutoScalingListenerForRunnerSet(ctx context.Context, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet, ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet, log logr.Logger) (ctrl.Result, error) {
+	imagePullSecrets := r.listenerImagePullSecrets()
+
 	r.ResourceCache.autoscalingListener.Delete(autoscalingRunnerSet)
 	autoscalingListener, err := r.newAutoscalingListener(
 		autoscalingRunnerSet,

controllers/actions.github.com/autoscalingrunnerset_controller_test.go

@@ -77,6 +77,12 @@ var _ = Describe("Test AutoScalingRunnerSet controller", Ordered, func() {
 			Log:                                logf.Log,
 			ControllerNamespace:                autoscalingNS.Name,
 			DefaultRunnerScaleSetListenerImage: "ghcr.io/actions/arc",
+			// Configured rather than left empty so the suite exercises the
+			// listener spec the controller actually builds. These come from
+			// controller configuration, so a path that derives a desired
+			// listener without them reads its own listener as drifted and
+			// rebuilds it on every reconcile.
+			DefaultRunnerScaleSetListenerImagePullSecrets: []string{"dockerhub"},
 			ResourceBuilder: ResourceBuilder{
 				ResourceCache: resourceCache,
 				SecretResolver: secretresolver.New(mgr.GetClient(), scalefake.NewMultiClient(
@@ -1169,6 +1175,35 @@ var _ = Describe("Test AutoScalingRunnerSet controller", Ordered, func() {
 				autoscalingRunnerSetTestTimeout,
 				autoscalingRunnerSetTestInterval,
 			).Should(BeEquivalentTo("testgroup2"), "AutoScalingRunnerSet should have the runner group in its annotation")
+
+			// The listener name is a hash over the runner group, so renaming the
+			// group renames the listener. Every lookup is by that derived name,
+			// so the listener created under the previous name is invisible to
+			// the controller from here on: nothing deletes it while the scale
+			// set lives, and it keeps acquiring jobs alongside its replacement.
+			Eventually(
+				func(g Gomega) {
+					var listeners v1alpha1.AutoscalingListenerList
+					g.Expect(k8sClient.List(ctx, &listeners, client.InNamespace(autoscalingRunnerSet.Namespace))).To(Succeed())
+
+					var live []string
+					for _, listener := range listeners.Items {
+						if listener.Spec.AutoscalingRunnerSetName != autoscalingRunnerSet.Name ||
+							listener.Spec.AutoscalingRunnerSetNamespace != autoscalingRunnerSet.Namespace ||
+							!listener.DeletionTimestamp.IsZero() {
+							continue
+						}
+						live = append(live, listener.Name)
+					}
+
+					g.Expect(live).To(
+						ConsistOf(scaleSetListenerName(updated)),
+						"a renamed scale set should be left with exactly one listener, under the current name",
+					)
+				},
+				autoscalingRunnerSetTestTimeout,
+				autoscalingRunnerSetTestInterval,
+			).Should(Succeed())
 		})
 	})
 
@@ -2879,8 +2914,16 @@ var _ = Describe("Test AutoscalingRunnerSet outdated lifecycle", Ordered, func()
 			return client.ObjectKey{Name: autoscalingRunnerSet.Name, Namespace: autoscalingRunnerSet.Namespace}
 		}
 
+		// The listener name is derived from the scale set as it stands now, not
+		// as the test declared it: the name is a hash over the runner group, so
+		// an edit to the group renames the listener the scale set should have.
 		listenerKey := func() client.ObjectKey {
-			return client.ObjectKey{Name: scaleSetListenerName(autoscalingRunnerSet), Namespace: autoscalingRunnerSet.Namespace}
+			GinkgoHelper()
+
+			current := new(v1alpha1.AutoscalingRunnerSet)
+			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), current)).To(Succeed(), "failed to get the autoscaling runner set")
+
+			return client.ObjectKey{Name: scaleSetListenerName(current), Namespace: autoscalingRunnerSet.Namespace}
 		}
 
 		getEphemeralRunnerSet := func() *v1alpha1.EphemeralRunnerSet {
@@ -2927,12 +2970,14 @@ var _ = Describe("Test AutoscalingRunnerSet outdated lifecycle", Ordered, func()
 				Should(BeEquivalentTo(v1alpha1.AutoscalingRunnerSetPhaseOutdated), "the autoscaling runner set should report the outdated phase")
 
 			Eventually(
-				func() bool {
-					return errors.IsNotFound(k8sClient.Get(ctx, listenerKey(), new(v1alpha1.AutoscalingListener)))
+				func(g Gomega) {
+					listener := new(v1alpha1.AutoscalingListener)
+					g.Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed(), "the listener should be kept as the record of a scale set that can come back")
+					g.Expect(listener.Spec.Phase).To(Equal(v1alpha1.AutoscalingListenerPhaseStopped), "the listener should be stopped so no further jobs are acquired")
 				},
 				autoscalingRunnerSetTestTimeout,
 				autoscalingRunnerSetTestInterval,
-			).Should(BeTrue(), "the listener should be removed so no further jobs are acquired")
+			).Should(Succeed())
 
 			// The set is kept, not deleted: deleting it would make the controller
 			// rebuild it from the same rejected spec on the very next reconcile.
@@ -2960,6 +3005,33 @@ var _ = Describe("Test AutoscalingRunnerSet outdated lifecycle", Ordered, func()
 			return runnerSet.Spec.ActionableRevision
 		}
 
+		// expectStaysOutdated asserts that an edit did not resurrect the scale
+		// set. Consistently rather than Eventually: the failure being guarded
+		// against is a spurious transition out of the outdated phase, which a
+		// single sample taken at the wrong moment would miss entirely.
+		expectStaysOutdated := func(outdatedRevision int64) {
+			GinkgoHelper()
+
+			Consistently(
+				func(g Gomega) {
+					phase, err := autoscalingRunnerSetPhase()
+					g.Expect(err).NotTo(HaveOccurred())
+					g.Expect(phase).To(BeEquivalentTo(v1alpha1.AutoscalingRunnerSetPhaseOutdated), "an edit outside the runner spec must not move the scale set out of the outdated phase")
+
+					listener := new(v1alpha1.AutoscalingListener)
+					g.Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed(), "the listener must be kept, not deleted")
+					g.Expect(listener.Spec.Phase).To(Equal(v1alpha1.AutoscalingListenerPhaseStopped), "the listener must stay stopped so no further jobs are acquired")
+
+					current := getEphemeralRunnerSet()
+					g.Expect(current.Spec.ActionableRevision).To(Equal(outdatedRevision), "the rejected runner spec must not be retried")
+					g.Expect(current.Spec.Replicas).To(BeZero())
+					g.Expect(current.Spec.PatchID).To(BeZero())
+				},
+				3*time.Second,
+				autoscalingRunnerSetTestInterval,
+			).Should(Succeed())
+		}
+
 		expectRecovered := func(outdatedRevision int64) {
 			GinkgoHelper()
 
@@ -2976,12 +3048,14 @@ var _ = Describe("Test AutoscalingRunnerSet outdated lifecycle", Ordered, func()
 			).Should(BeNumerically(">", outdatedRevision), "the runner spec revision should advance so the runner set stops judging itself by the rejected runners")
 
 			Eventually(
-				func() error {
-					return k8sClient.Get(ctx, listenerKey(), new(v1alpha1.AutoscalingListener))
+				func(g Gomega) {
+					listener := new(v1alpha1.AutoscalingListener)
+					g.Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed())
+					g.Expect(listener.Spec.Phase.Stopped()).To(BeFalse(), "the listener should be started again so the scale set can acquire jobs")
 				},
 				autoscalingRunnerSetTestTimeout,
 				autoscalingRunnerSetTestInterval,
-			).Should(Succeed(), "the listener should be created again so the scale set can acquire jobs")
+			).Should(Succeed())
 
 			Eventually(autoscalingRunnerSetPhase, autoscalingRunnerSetTestTimeout, autoscalingRunnerSetTestInterval).
 				Should(BeEquivalentTo(v1alpha1.AutoscalingRunnerSetPhaseRunning), "the autoscaling runner set should leave the outdated phase")
@@ -3073,12 +3147,98 @@ var _ = Describe("Test AutoscalingRunnerSet outdated lifecycle", Ordered, func()
 			expectRecovered(outdatedRevision)
 		})
 
-		// The runner spec is not the only reason a scale set can be stuck: the
-		// runners may have been rejected because of the scale set registration
-		// rather than the pod template. Any spec edit therefore has to be enough
-		// to retry, otherwise the scale set can only be recovered by touching a
-		// field that has nothing to do with the failure.
-		It("recovers when a field outside the runner spec is updated", func() {
+		// While a scale set is parked no listener spec drift is propagated, but
+		// edits outside the runner spec are still accepted and recorded. The
+		// listener that comes back on recovery therefore has to carry them.
+		//
+		// This covers the end state only. The transient it guards against - a
+		// listener started from the spec it was parked with, running under stale
+		// configuration until the next reconcile replaces it - is too short to
+		// observe here, and is covered deterministically by
+		// TestAutoscalingRunnerSetReplacesAStoppedListenerWithADriftedSpec.
+		It("propagates an edit made while outdated and keeps it on recovery", func() {
+			markRunnersOutdated()
+			outdatedRevision := expectSwitchedOff()
+
+			updated := new(v1alpha1.AutoscalingRunnerSet)
+			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), updated)).To(Succeed())
+			original := updated.DeepCopy()
+			max := 20
+			updated.Spec.MaxRunners = &max
+			Expect(k8sClient.Patch(ctx, updated, client.MergeFrom(original))).To(Succeed(), "failed to update the replica bounds")
+
+			expectStaysOutdated(outdatedRevision)
+
+			// Switched off is not frozen: the edit lands on the listener while it
+			// is still stopped, rather than waiting for the scale set to recover.
+			Eventually(
+				func(g Gomega) {
+					listener := new(v1alpha1.AutoscalingListener)
+					g.Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed())
+					g.Expect(listener.Spec.Phase).To(Equal(v1alpha1.AutoscalingListenerPhaseStopped), "the listener must stay switched off")
+					g.Expect(listener.Spec.MaxRunners).To(Equal(max), "the edit should reach the listener while it is stopped")
+				},
+				autoscalingRunnerSetTestTimeout,
+				autoscalingRunnerSetTestInterval,
+			).Should(Succeed())
+
+			updated = new(v1alpha1.AutoscalingRunnerSet)
+			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), updated)).To(Succeed())
+			original = updated.DeepCopy()
+			updated.Spec.Template.Spec.Containers[0].Image = "ghcr.io/actions/runner:fixed"
+			Expect(k8sClient.Patch(ctx, updated, client.MergeFrom(original))).To(Succeed(), "failed to correct the runner spec")
+
+			expectRecovered(outdatedRevision)
+
+			Eventually(
+				func(g Gomega) {
+					listener := new(v1alpha1.AutoscalingListener)
+					g.Expect(k8sClient.Get(ctx, listenerKey(), listener)).To(Succeed())
+					g.Expect(listener.Spec.Phase.Stopped()).To(BeFalse())
+					g.Expect(listener.Spec.MaxRunners).To(Equal(max), "the edit taken while the scale set was parked should reach the listener on recovery")
+				},
+				autoscalingRunnerSetTestTimeout,
+				autoscalingRunnerSetTestInterval,
+			).Should(Succeed())
+		})
+
+		// The runner spec is not the only part of the EphemeralRunnerSet spec the
+		// AutoscalingRunnerSet owns: the metadata stamped onto the runners it
+		// creates is published the same way and changes what the next runner
+		// looks like. It therefore recovers the scale set too.
+		It("recovers when the runner metadata is corrected", func() {
+			markRunnersOutdated()
+			outdatedRevision := expectSwitchedOff()
+
+			updated := new(v1alpha1.AutoscalingRunnerSet)
+			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), updated)).To(Succeed())
+			original := updated.DeepCopy()
+			updated.Spec.EphemeralRunnerMetadata = &v1alpha1.ResourceMeta{
+				Labels: map[string]string{"arc.test/runner": "corrected"},
+			}
+			Expect(k8sClient.Patch(ctx, updated, client.MergeFrom(original))).To(Succeed(), "failed to correct the runner metadata")
+
+			expectRecovered(outdatedRevision)
+
+			Eventually(
+				func() map[string]string {
+					current := getEphemeralRunnerSet()
+					if current.Spec.EphemeralRunnerMetadata == nil {
+						return nil
+					}
+					return current.Spec.EphemeralRunnerMetadata.Labels
+				},
+				autoscalingRunnerSetTestTimeout,
+				autoscalingRunnerSetTestInterval,
+			).Should(HaveKeyWithValue("arc.test/runner", "corrected"), "the corrected runner metadata should be published to the set")
+		})
+
+		// The outdated phase is sticky. Only a change to what the runners would
+		// be handed next - the runner spec or the metadata stamped onto them -
+		// is evidence that retrying is worth anything. Any other edit would
+		// otherwise switch the listener back on and start acquiring jobs against
+		// runners that will reject the spec exactly as before.
+		It("stays outdated when the replica bounds are updated", func() {
 			markRunnersOutdated()
 			outdatedRevision := expectSwitchedOff()
 
@@ -3089,7 +3249,33 @@ var _ = Describe("Test AutoscalingRunnerSet outdated lifecycle", Ordered, func()
 			updated.Spec.MaxRunners = &max
 			Expect(k8sClient.Patch(ctx, updated, client.MergeFrom(original))).To(Succeed(), "failed to update the autoscaling runner set")
 
-			expectRecovered(outdatedRevision)
+			expectStaysOutdated(outdatedRevision)
+		})
+
+		It("stays outdated when the runner group is updated", func() {
+			markRunnersOutdated()
+			outdatedRevision := expectSwitchedOff()
+
+			updated := new(v1alpha1.AutoscalingRunnerSet)
+			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), updated)).To(Succeed())
+			original := updated.DeepCopy()
+			updated.Spec.RunnerGroup = "othergroup"
+			Expect(k8sClient.Patch(ctx, updated, client.MergeFrom(original))).To(Succeed(), "failed to update the runner group")
+
+			expectStaysOutdated(outdatedRevision)
+		})
+
+		It("stays outdated when the runner scale set name is updated", func() {
+			markRunnersOutdated()
+			outdatedRevision := expectSwitchedOff()
+
+			updated := new(v1alpha1.AutoscalingRunnerSet)
+			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), updated)).To(Succeed())
+			original := updated.DeepCopy()
+			updated.Spec.RunnerScaleSetName = "renamed-scale-set"
+			Expect(k8sClient.Patch(ctx, updated, client.MergeFrom(original))).To(Succeed(), "failed to update the runner scale set name")
+
+			expectStaysOutdated(outdatedRevision)
 		})
 
 		It("does not retry an outdated runner spec during a metadata-only listener rebuild", func() {

controllers/actions.github.com/autoscalingrunnerset_listener_cache_test.go

@@ -0,0 +1,67 @@
+package actionsgithubcom
+
+import (
+	"context"
+	"testing"
+
+	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// newAutoscalingListener serves a shared pointer out of the resource cache, so a
+// caller that stamps the stopped phase onto what it gets back stamps it onto
+// every later caller's desired listener too. That would eventually create a
+// listener that is born stopped and never starts.
+func TestCreatingAStoppedListenerDoesNotPoisonTheResourceCache(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	ctx := context.Background()
+
+	ephemeralRunnerSet := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), ephemeralRunnerSet))
+	// The cache is keyed on the UID, and the fake client does not assign one.
+	autoscalingRunnerSet.UID = "ars-uid"
+
+	first, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		reconciler.DefaultRunnerScaleSetListenerImage,
+		reconciler.listenerImagePullSecrets(),
+	)
+	require.NoError(t, err)
+	cached, hit := reconciler.ResourceCache.autoscalingListener.Get(
+		autoscalingRunnerSet,
+		first,
+		ephemeralRunnerSet,
+		resourceCacheInputObject("autoscaling-listener-inputs", struct {
+			Namespace        string
+			Image            string
+			ImagePullSecrets []corev1.LocalObjectReference
+		}{
+			Namespace:        reconciler.ControllerNamespace,
+			Image:            reconciler.DefaultRunnerScaleSetListenerImage,
+			ImagePullSecrets: reconciler.listenerImagePullSecrets(),
+		}),
+	)
+	require.True(t, hit, "this test is only meaningful when the cache is actually serving the listener")
+	require.Same(t, first, cached, "the cache hands out a shared pointer")
+
+	require.NoError(t, reconciler.createStoppedListener(ctx, autoscalingRunnerSet, ephemeralRunnerSet, logr.Discard()))
+
+	again, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		reconciler.DefaultRunnerScaleSetListenerImage,
+		reconciler.listenerImagePullSecrets(),
+	)
+	require.NoError(t, err)
+	require.False(
+		t,
+		again.Spec.Phase.Stopped(),
+		"the desired listener every other caller derives must not have been switched off",
+	)
+}

controllers/actions.github.com/autoscalingrunnerset_outdated_recovery_test.go

@@ -24,32 +24,52 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
 	"github.com/actions/actions-runner-controller/build"
 )
 
-func TestAutoscalingRunnerSetParksFirstRejectionOfPublishedGeneration(t *testing.T) {
+// outdatedFixture builds an AutoscalingRunnerSet whose EphemeralRunnerSet has
+// reported that its runners rejected the runner spec, together with a client and
+// a reconciler wired to them.
+// publishedGeneration is the AutoscalingRunnerSet generation recorded on the
+// EphemeralRunnerSet. Lagging behind the live generation is what a
+// generation-based recovery signal reads as "retry the spec the runners
+// rejected"; matching it is what such a signal reads as "never retry", even when
+// the runner spec itself has changed.
+func outdatedFixture(
+	t *testing.T,
+	runnerSetPhase v1alpha1.AutoscalingRunnerSetPhase,
+	generation int64,
+	publishedGeneration int64,
+	desiredImage string,
+) (*v1alpha1.AutoscalingRunnerSet, *AutoscalingRunnerSetReconciler, client.Client) {
+	t.Helper()
+
 	scheme := runtime.NewScheme()
 	require.NoError(t, corev1.AddToScheme(scheme))
 	require.NoError(t, v1alpha1.AddToScheme(scheme))
 
 	const (
-		name       = "test"
-		namespace  = "test"
-		generation = int64(2)
-		revision   = int64(1)
+		name      = "test"
+		namespace = "test"
 	)
-	template := corev1.PodTemplateSpec{
+
+	rejectedTemplate := corev1.PodTemplateSpec{
 		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{{Name: "runner", Image: "runner:new"}},
+			Containers: []corev1.Container{{Name: "runner", Image: "runner:rejected"}},
 		},
 	}
+	desiredTemplate := *rejectedTemplate.DeepCopy()
+	desiredTemplate.Spec.Containers[0].Image = desiredImage
+
 	autoscalingRunnerSet := &v1alpha1.AutoscalingRunnerSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:       name,
@@ -65,10 +85,10 @@ func TestAutoscalingRunnerSetParksFirstRejectionOfPublishedGeneration(t *testing
 		},
 		Spec: v1alpha1.AutoscalingRunnerSetSpec{
 			GitHubConfigUrl: "https://github.com/owner/repo",
-			Template:        template,
+			Template:        desiredTemplate,
 		},
 		Status: v1alpha1.AutoscalingRunnerSetStatus{
-			Phase:              v1alpha1.AutoscalingRunnerSetPhasePending,
+			Phase:              runnerSetPhase,
 			ObservedGeneration: generation - 1,
 		},
 	}
@@ -77,27 +97,29 @@ func TestAutoscalingRunnerSetParksFirstRejectionOfPublishedGeneration(t *testing
 			Name:      name,
 			Namespace: namespace,
 			Annotations: map[string]string{
-				AnnotationKeyAutoscalingRunnerSetGeneration: strconv.FormatInt(generation, 10),
+				AnnotationKeyAutoscalingRunnerSetGeneration: strconv.FormatInt(publishedGeneration, 10),
 			},
 		},
 		Spec: v1alpha1.EphemeralRunnerSetSpec{
 			Replicas:           3,
 			PatchID:            7,
-			ActionableRevision: revision,
+			ActionableRevision: outdatedFixtureRevision,
 			EphemeralRunnerSpec: v1alpha1.EphemeralRunnerSpec{
 				RunnerScaleSetID: 1,
 				GitHubConfigURL:  autoscalingRunnerSet.Spec.GitHubConfigUrl,
-				PodTemplateSpec:  template,
+				PodTemplateSpec:  rejectedTemplate,
 			},
 		},
 		Status: v1alpha1.EphemeralRunnerSetStatus{
 			Phase:                     v1alpha1.EphemeralRunnerSetPhaseOutdated,
-			AppliedActionableRevision: revision,
+			AppliedActionableRevision: outdatedFixtureRevision,
 		},
 	}
 
 	c := fake.NewClientBuilder().
 		WithScheme(scheme).
+		// SetupIndexers registers this on the manager's cache.
+		WithIndex(&v1alpha1.AutoscalingListener{}, autoscalingRunnerSetOwnerKey, autoscalingListenerRunnerSetIndexer).
 		WithStatusSubresource(autoscalingRunnerSet, ephemeralRunnerSet).
 		WithObjects(autoscalingRunnerSet, ephemeralRunnerSet).
 		Build()
@@ -113,18 +135,433 @@ func TestAutoscalingRunnerSetParksFirstRejectionOfPublishedGeneration(t *testing
 		},
 	}
 
-	_, err := reconciler.Reconcile(context.Background(), ctrl.Request{
-		NamespacedName: types.NamespacedName{Name: name, Namespace: namespace},
-	})
+	return autoscalingRunnerSet, reconciler, c
+}
+
+const outdatedFixtureRevision = int64(1)
+
+func reconcileOutdatedFixture(t *testing.T, reconciler *AutoscalingRunnerSetReconciler, key types.NamespacedName) {
+	t.Helper()
+
+	_, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: key})
 	require.NoError(t, err)
+}
+
+// The set is switched off on the first reconcile that sees the rejection, even
+// though the AutoscalingRunnerSet still has an unobserved generation. The
+// generation says nothing about the spec the runners objected to.
+func TestAutoscalingRunnerSetParksFirstRejection(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhasePending, 2, 1, "runner:rejected")
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+
+	reconcileOutdatedFixture(t, reconciler, key)
 
 	gotARS := new(v1alpha1.AutoscalingRunnerSet)
-	require.NoError(t, c.Get(context.Background(), types.NamespacedName{Name: name, Namespace: namespace}, gotARS))
+	require.NoError(t, c.Get(context.Background(), key, gotARS))
 	require.Equal(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, gotARS.Status.Phase)
 
 	gotERS := new(v1alpha1.EphemeralRunnerSet)
-	require.NoError(t, c.Get(context.Background(), types.NamespacedName{Name: name, Namespace: namespace}, gotERS))
-	require.Equal(t, revision, gotERS.Spec.ActionableRevision)
+	require.NoError(t, c.Get(context.Background(), key, gotERS))
+	require.Equal(t, outdatedFixtureRevision, gotERS.Spec.ActionableRevision)
 	require.Zero(t, gotERS.Spec.Replicas)
 	require.Zero(t, gotERS.Spec.PatchID)
 }
+
+// An already outdated scale set stays outdated while the runner spec is the one
+// that was rejected, no matter how far its generation has moved on.
+func TestAutoscalingRunnerSetStaysOutdatedWithoutRunnerSpecChange(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+
+	reconcileOutdatedFixture(t, reconciler, key)
+
+	gotARS := new(v1alpha1.AutoscalingRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotARS))
+	require.Equal(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, gotARS.Status.Phase)
+
+	gotERS := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotERS))
+	require.Equal(t, outdatedFixtureRevision, gotERS.Spec.ActionableRevision, "the rejected runner spec must not be retried")
+
+	// The fixture carries a nonzero target, which is what a listener that was
+	// still draining when the set was parked would have left behind. Re-pinning
+	// it is what makes that harmless: the AutoscalingRunnerSet owns the
+	// EphemeralRunnerSet, so a target published while parked re-enqueues it and
+	// is taken straight back to zero.
+	require.Zero(t, gotERS.Spec.Replicas, "a target published while parked must be taken back to zero")
+	require.Zero(t, gotERS.Spec.PatchID)
+}
+
+// Correcting the runner spec is the one edit that recovers the scale set: the
+// phase leaves outdated and the new spec is published with a higher revision, so
+// the EphemeralRunnerSet stops judging itself by the runners that failed.
+func TestAutoscalingRunnerSetRecoversOnRunnerSpecChange(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:fixed")
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+
+	reconcileOutdatedFixture(t, reconciler, key)
+
+	gotARS := new(v1alpha1.AutoscalingRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotARS))
+	require.Equal(t, v1alpha1.AutoscalingRunnerSetPhasePending, gotARS.Status.Phase)
+
+	gotERS := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotERS))
+	require.Equal(t, "runner:fixed", gotERS.Spec.EphemeralRunnerSpec.Spec.Containers[0].Image)
+	require.Greater(t, gotERS.Spec.ActionableRevision, outdatedFixtureRevision, "the revision must advance so the failed runners are treated as stale")
+}
+
+// The mirror image of over-parking: the runner spec changed, but the generation
+// recorded on the EphemeralRunnerSet is already current, which is what happens
+// whenever the derived runner spec moves without metadata.generation moving with
+// it. A generation-based signal refuses the recovery; comparing the spec does
+// not.
+func TestAutoscalingRunnerSetRecoversWithoutAnUnobservedGeneration(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 5, "runner:fixed")
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+
+	reconcileOutdatedFixture(t, reconciler, key)
+
+	gotARS := new(v1alpha1.AutoscalingRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotARS))
+	require.Equal(t, v1alpha1.AutoscalingRunnerSetPhasePending, gotARS.Status.Phase)
+
+	gotERS := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotERS))
+	require.Equal(t, "runner:fixed", gotERS.Spec.EphemeralRunnerSpec.Spec.Containers[0].Image)
+	require.Greater(t, gotERS.Spec.ActionableRevision, outdatedFixtureRevision)
+}
+
+// Runner metadata is published to the set the same way the runner spec is, and
+// changes what the next runner looks like, so it recovers the scale set too. The
+// revision must advance with it: without that the EphemeralRunnerSet would keep
+// judging itself by the runners that failed and push the scale set straight back
+// to outdated.
+func TestAutoscalingRunnerSetRecoversOnRunnerMetadataChange(t *testing.T) {
+	// Published generation deliberately current, so the recovery can only come
+	// from the metadata comparison and not from a generation that lags.
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 5, "runner:rejected")
+	autoscalingRunnerSet.Spec.EphemeralRunnerMetadata = &v1alpha1.ResourceMeta{
+		Labels: map[string]string{"arc.test/runner": "corrected"},
+	}
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+	require.NoError(t, c.Update(context.Background(), autoscalingRunnerSet))
+
+	reconcileOutdatedFixture(t, reconciler, key)
+
+	gotARS := new(v1alpha1.AutoscalingRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotARS))
+	require.Equal(t, v1alpha1.AutoscalingRunnerSetPhasePending, gotARS.Status.Phase)
+
+	gotERS := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(context.Background(), key, gotERS))
+	require.NotNil(t, gotERS.Spec.EphemeralRunnerMetadata)
+	require.Equal(t, "corrected", gotERS.Spec.EphemeralRunnerMetadata.Labels["arc.test/runner"])
+	require.Greater(t, gotERS.Spec.ActionableRevision, outdatedFixtureRevision, "the revision must advance so the failed runners are treated as stale")
+}
+
+// A parked scale set still accepts edits that are not a recovery signal, and
+// reconcileOutdated returns before any of them reach the listener. Starting the
+// listener again therefore has to reckon with a spec that has moved on since it
+// was stopped: it is replaced, not simply switched back on, so it never runs for
+// a moment under the configuration the scale set was parked with.
+func TestAutoscalingRunnerSetReplacesAStoppedListenerWithADriftedSpec(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseRunning, 5, 5, "runner:rejected")
+	ctx := context.Background()
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+
+	// The edit the scale set took while it was parked. It is not a runner spec
+	// change, so it never propagated to the listener.
+	maxRunners := 20
+	autoscalingRunnerSet.Spec.MaxRunners = &maxRunners
+	autoscalingRunnerSet.Status.ObservedGeneration = autoscalingRunnerSet.Generation
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+	require.NoError(t, c.Status().Update(ctx, autoscalingRunnerSet))
+
+	// The set is settled on the current spec and no longer complaining, so this
+	// reconcile has nothing to publish to it and reaches the listener.
+	desiredERS, err := reconciler.newEphemeralRunnerSet(autoscalingRunnerSet)
+	require.NoError(t, err)
+	gotERS := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, key, gotERS))
+	gotERS.Spec = desiredERS.Spec
+	gotERS.Labels = desiredERS.Labels
+	gotERS.Annotations = desiredERS.Annotations
+	require.NoError(t, c.Update(ctx, gotERS))
+	require.NoError(t, c.Get(ctx, key, gotERS))
+	gotERS.Status.Phase = v1alpha1.EphemeralRunnerSetPhaseRunning
+	require.NoError(t, c.Status().Update(ctx, gotERS))
+
+	parked := autoscalingRunnerSet.DeepCopy()
+	parked.Spec.MaxRunners = nil
+	listener, err := reconciler.newAutoscalingListener(parked, gotERS, reconciler.ControllerNamespace, "listener:image", nil)
+	require.NoError(t, err)
+	listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+	require.NoError(t, c.Create(ctx, listener))
+
+	reconcileOutdatedFixture(t, reconciler, key)
+
+	err = c.Get(ctx, client.ObjectKeyFromObject(listener), new(v1alpha1.AutoscalingListener))
+	require.True(
+		t,
+		kerrors.IsNotFound(err),
+		"a stopped listener whose spec has drifted must be replaced rather than started, so it is never running with the spec the scale set was parked with",
+	)
+}
+
+// A parked scale set still accepts edits that are not a recovery signal. They
+// must reach the objects they belong to: the scale set stays switched off, but
+// switched off is not the same as frozen, and an edit that silently never lands
+// would come back as a surprise whenever the set eventually recovers.
+func TestAutoscalingRunnerSetPropagatesUnrelatedEditsWhileOutdated(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	ctx := context.Background()
+	key := client.ObjectKeyFromObject(autoscalingRunnerSet)
+
+	// Neither edit touches the runner spec, so neither recovers the scale set.
+	maxRunners := 20
+	autoscalingRunnerSet.Spec.MaxRunners = &maxRunners
+	autoscalingRunnerSet.Labels["arc.test/edited-while-parked"] = "yes"
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+
+	listener, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		&v1alpha1.EphemeralRunnerSet{ObjectMeta: metav1.ObjectMeta{Name: autoscalingRunnerSet.Name, Namespace: autoscalingRunnerSet.Namespace}},
+		reconciler.ControllerNamespace,
+		"listener:image",
+		nil,
+	)
+	require.NoError(t, err)
+	// The listener as it was when the scale set was parked: no max runners.
+	listener.Spec.MaxRunners = 0
+	listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+	require.NoError(t, c.Create(ctx, listener))
+
+	// More than one reconcile, because the edits land on different objects and
+	// the controller returns after each patch.
+	for range 4 {
+		reconcileOutdatedFixture(t, reconciler, key)
+	}
+
+	gotARS := new(v1alpha1.AutoscalingRunnerSet)
+	require.NoError(t, c.Get(ctx, key, gotARS))
+	require.Equal(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, gotARS.Status.Phase, "an edit outside the runner spec must not un-park the scale set")
+
+	gotListener := new(v1alpha1.AutoscalingListener)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(listener), gotListener))
+	require.Equal(t, v1alpha1.AutoscalingListenerPhaseStopped, gotListener.Spec.Phase, "the listener must stay switched off")
+	require.Equal(t, maxRunners, gotListener.Spec.MaxRunners, "the edit must reach the listener even though it is switched off")
+
+	gotERS := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, key, gotERS))
+	require.Equal(t, "yes", gotERS.Labels["arc.test/edited-while-parked"], "the edit must reach the ephemeral runner set even though it is switched off")
+	require.Equal(t, outdatedFixtureRevision, gotERS.Spec.ActionableRevision, "the rejected runner spec must not be retried")
+	require.Zero(t, gotERS.Spec.Replicas, "the set must stay pinned at zero replicas")
+	require.Zero(t, gotERS.Spec.PatchID)
+}
+
+// propagateToStoppedListener runs immediately after stopListener has patched the
+// phase, and reads the listener back through the cache. That read can still hold
+// the pre-patch object, so the live phase is not a value this helper can trust:
+// carrying it over would write Running back onto a listener the same reconcile
+// has just switched off, while the scale set stays outdated.
+//
+// The helper only ever runs on the parked path, so the phase it writes is not in
+// question. It is Stopped.
+func TestPropagateToStoppedListenerNeverRestartsTheListener(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	ctx := context.Background()
+
+	maxRunners := 20
+	autoscalingRunnerSet.Spec.MaxRunners = &maxRunners
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+
+	ephemeralRunnerSet := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), ephemeralRunnerSet))
+
+	listener, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		"listener:image",
+		nil,
+	)
+	require.NoError(t, err)
+	// The listener as a stale cached read returns it: still running, and still
+	// carrying the spec it was parked with, so there is drift to propagate.
+	listener.Spec.MaxRunners = 0
+	listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseRunning
+	require.NoError(t, c.Create(ctx, listener))
+
+	require.NoError(t, reconciler.propagateToStoppedListener(ctx, autoscalingRunnerSet, ephemeralRunnerSet, logr.Discard()))
+
+	got := new(v1alpha1.AutoscalingListener)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(listener), got))
+	require.Equal(t, maxRunners, got.Spec.MaxRunners, "the edit should still be propagated")
+	require.True(
+		t,
+		got.Spec.Phase.Stopped(),
+		"propagating an edit must never restart a listener the same reconcile switched off, whatever phase the cached read reported",
+	)
+}
+
+// A parked scale set now accepts edits, and the listener's name is a hash over
+// the runner group and the config URL, so an edit to either renames the listener
+// the controller looks for. Every lookup is by that derived name, so the running
+// listener created under the old name is invisible to the parked path: it would
+// keep acquiring jobs for a scale set that is supposed to be switched off.
+func TestStopListenerSwitchesOffAListenerThatTheRunnerGroupRenamed(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	ctx := context.Background()
+
+	ephemeralRunnerSet := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), ephemeralRunnerSet))
+
+	listener, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		"listener:image",
+		nil,
+	)
+	require.NoError(t, err)
+	require.NoError(t, c.Create(ctx, listener))
+
+	autoscalingRunnerSet.Spec.RunnerGroup = "moved-to-another-group"
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+	require.NotEqual(
+		t,
+		listener.Name,
+		scaleSetListenerName(autoscalingRunnerSet),
+		"this test is only meaningful while the runner group renames the listener",
+	)
+
+	require.NoError(t, reconciler.stopListener(ctx, autoscalingRunnerSet, logr.Discard()))
+
+	var listeners v1alpha1.AutoscalingListenerList
+	require.NoError(t, c.List(ctx, &listeners, client.InNamespace(reconciler.ControllerNamespace)))
+	for _, got := range listeners.Items {
+		require.True(
+			t,
+			got.Spec.Phase.Stopped(),
+			"listener %q kept acquiring jobs for a switched-off scale set",
+			got.Name,
+		)
+	}
+}
+
+// A listener left behind under a previous name can never be started again: every
+// path that starts one looks it up by the name derived now. Keeping it in step
+// with the desired spec would only maintain a permanent orphan, so the parked
+// path deletes it instead, and recovery builds the listener fresh.
+func TestPropagateToStoppedListenerDeletesARenamedListenerRatherThanUpdatingIt(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	ctx := context.Background()
+
+	ephemeralRunnerSet := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), ephemeralRunnerSet))
+
+	listener, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		"listener:image",
+		nil,
+	)
+	require.NoError(t, err)
+	listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+	require.NoError(t, c.Create(ctx, listener))
+
+	autoscalingRunnerSet.Spec.RunnerGroup = "moved-to-another-group"
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+	require.NotEqual(
+		t,
+		listener.Name,
+		scaleSetListenerName(autoscalingRunnerSet),
+		"this test is only meaningful while the runner group renames the listener",
+	)
+
+	require.NoError(t, reconciler.propagateToStoppedListener(ctx, autoscalingRunnerSet, ephemeralRunnerSet, logr.Discard()))
+
+	err = c.Get(ctx, client.ObjectKeyFromObject(listener), new(v1alpha1.AutoscalingListener))
+	require.True(t, kerrors.IsNotFound(err), "the listener under the previous name should be gone, got %v", err)
+}
+
+// Teardown is gated on the listener being gone, and reporting that while one
+// still exists lets the AutoscalingRunnerSet's finalizer be removed out from
+// under a listener that still has a pod and is still acquiring jobs.
+func TestCleanupListenerWaitsForAListenerLeftUnderAPreviousName(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseRunning, 5, 5, "runner:rejected")
+	ctx := context.Background()
+
+	ephemeralRunnerSet := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), ephemeralRunnerSet))
+
+	listener, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		"listener:image",
+		nil,
+	)
+	require.NoError(t, err)
+	require.NoError(t, c.Create(ctx, listener))
+
+	autoscalingRunnerSet.Spec.RunnerGroup = "moved-to-another-group"
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+	require.NotEqual(
+		t,
+		listener.Name,
+		scaleSetListenerName(autoscalingRunnerSet),
+		"this test is only meaningful while the runner group renames the listener",
+	)
+
+	done, err := reconciler.cleanupListener(ctx, autoscalingRunnerSet, logr.Discard())
+	require.NoError(t, err)
+	require.False(t, done, "teardown must not report the listener gone while one still exists under a previous name")
+
+	err = c.Get(ctx, client.ObjectKeyFromObject(listener), new(v1alpha1.AutoscalingListener))
+	require.True(t, kerrors.IsNotFound(err), "the listener should have been deleted, got %v", err)
+}
+
+// The listener spec the parked path derives has to be the same one creation
+// derives, or propagating an unrelated edit silently rewrites the fields the
+// parked path does not know about. Image pull secrets are the ones that bite:
+// they come from controller configuration rather than from the
+// AutoscalingRunnerSet, so a listener parked with credentials for a private
+// image would come back without them and fail to pull.
+func TestPropagateToStoppedListenerKeepsTheConfiguredImagePullSecrets(t *testing.T) {
+	autoscalingRunnerSet, reconciler, c := outdatedFixture(t, v1alpha1.AutoscalingRunnerSetPhaseOutdated, 5, 4, "runner:rejected")
+	reconciler.DefaultRunnerScaleSetListenerImagePullSecrets = []string{"private-registry"}
+	ctx := context.Background()
+
+	ephemeralRunnerSet := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(autoscalingRunnerSet), ephemeralRunnerSet))
+
+	listener, err := reconciler.newAutoscalingListener(
+		autoscalingRunnerSet,
+		ephemeralRunnerSet,
+		reconciler.ControllerNamespace,
+		"listener:image",
+		[]corev1.LocalObjectReference{{Name: "private-registry"}},
+	)
+	require.NoError(t, err)
+	listener.Spec.Phase = v1alpha1.AutoscalingListenerPhaseStopped
+	require.NoError(t, c.Create(ctx, listener))
+
+	maxRunners := 20
+	autoscalingRunnerSet.Spec.MaxRunners = &maxRunners
+	require.NoError(t, c.Update(ctx, autoscalingRunnerSet))
+
+	require.NoError(t, reconciler.propagateToStoppedListener(ctx, autoscalingRunnerSet, ephemeralRunnerSet, logr.Discard()))
+
+	got := new(v1alpha1.AutoscalingListener)
+	require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(listener), got))
+	require.Equal(t, maxRunners, got.Spec.MaxRunners, "the edit should still be propagated")
+	require.Equal(
+		t,
+		[]corev1.LocalObjectReference{{Name: "private-registry"}},
+		got.Spec.ImagePullSecrets,
+		"propagating an unrelated edit must not drop the credentials the listener image is pulled with",
+	)
+}

controllers/actions.github.com/constants.go

@@ -51,9 +51,11 @@ const (
 	AnnotationKeyGitHubRunnerScaleSetName = "actions.github.com/runner-scale-set-name"
 	AnnotationKeyPatchID                  = "actions.github.com/patch-id"
 	// AnnotationKeyAutoscalingRunnerSetGeneration records the AutoscalingRunnerSet
-	// generation that published the current EphemeralRunnerSet actionable
-	// revision. It prevents a rejected revision from being retried more than once
-	// for the same AutoscalingRunnerSet spec update.
+	// generation the current EphemeralRunnerSet spec was derived from. It is
+	// informational: it makes it possible to tell, by looking at the set alone,
+	// how far behind the AutoscalingRunnerSet it is. Nothing keys behaviour off
+	// it - in particular, recovery from the outdated phase is decided by
+	// comparing the runner spec itself, not generations.
 	AnnotationKeyAutoscalingRunnerSetGeneration = "actions.github.com/autoscaling-runner-set-generation"
 	// AnnotationKeyActionableRevision records the EphemeralRunnerSet
 	// Spec.ActionableRevision that was in effect when the runner was created. It
@@ -96,6 +98,12 @@ const DefaultScaleSetListenerLogFormat = string(logging.LogFormatText)
 // ownerKey is field selector matching the owner name of a particular resource
 const resourceOwnerKey = ".metadata.controller"
 
+// autoscalingRunnerSetOwnerKey indexes an AutoscalingListener by the scale set
+// it names as its own. Listeners live in the controller namespace while the
+// scale set lives in its own, so they cannot carry an owner reference across
+// that boundary and the resourceOwnerKey index does not apply to them.
+const autoscalingRunnerSetOwnerKey = ".spec.autoscalingRunnerSet"
+
 // EphemeralRunner pod creation failure reasons
 const (
 	ReasonTooManyPodFailures = "TooManyPodFailures"

controllers/actions.github.com/ephemeralrunnerset_controller.go

@@ -229,6 +229,43 @@ func (r *EphemeralRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl.R
 		return ctrl.Result{}, r.updateStatus(ctx, &ephemeralRunnerSet, ephemeralRunnersByState, log)
 	}
 
+	// A runner that rejected the spec it was given settles the question the
+	// target count was asking. Spec.Replicas is what the listener wanted,
+	// computed from queued jobs before anything was known to be wrong with the
+	// spec those runners would be built from; acting on it now would create
+	// runners that reject it in exactly the same way. Release what can be
+	// released and record the rejection, rather than falling through to the
+	// scaling block below.
+	//
+	// The recorded phase is not enough to enforce this on its own. It only turns
+	// Outdated at the end of this reconcile, so the pass that discovers the
+	// rejection would otherwise scale up against a spec already known to be bad,
+	// and only the pass after it would take the early return above.
+	//
+	// The phase is recorded before the runners are released, not after. The
+	// cleanup deletes the outdated runners themselves, so a failure part way
+	// through would otherwise leave a set with no recorded rejection and fewer
+	// runners to rediscover it from. With the phase written first, the early
+	// return above picks the work up instead.
+	if len(ephemeralRunnersByState.outdated) > 0 {
+		log.Info(
+			"Ephemeral runners rejected the runner spec. Releasing runners instead of applying the target count",
+			"outdated", len(ephemeralRunnersByState.outdated),
+			"desired", ephemeralRunnerSet.Spec.Replicas,
+		)
+		if err := r.updateStatus(ctx, &ephemeralRunnerSet, ephemeralRunnersByState, log); err != nil {
+			log.Error(err, "Failed to record the outdated phase")
+			return ctrl.Result{}, err
+		}
+
+		if _, err := r.cleanUpEphemeralRunners(ctx, &ephemeralRunnerSet, log); err != nil {
+			log.Error(err, "Failed to clean up EphemeralRunners")
+			return ctrl.Result{}, err
+		}
+
+		return ctrl.Result{}, nil
+	}
+
 	total := ephemeralRunnersByState.scaleTotal()
 	if ephemeralRunnerSet.Spec.PatchID == 0 || ephemeralRunnerSet.Spec.PatchID != ephemeralRunnersByState.latestPatchID {
 		// Spec.Replicas is the count the listener asked for when it published

controllers/actions.github.com/ephemeralrunnerset_outdated_target_test.go

@@ -0,0 +1,135 @@
+package actionsgithubcom
+
+import (
+	"context"
+	"testing"
+
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
+)
+
+// TestEphemeralRunnerSetIgnoresTheTargetCountOnTheReconcileThatFindsOutdatedRunners
+// pins that a rejected runner spec beats the count the listener asked for.
+//
+// Spec.Replicas is what the listener wanted, computed before any runner had
+// reported anything. Once a runner rejects the spec it was given, that count is
+// a request to create more runners that will reject it in exactly the same way,
+// so it must not be acted on.
+//
+// The phase alone is not enough to enforce that. Status.Phase only turns
+// Outdated at the end of the reconcile that discovers the rejection, so the
+// reconcile that finds it first still reaches the scaling block below with a
+// phase that still says Running, and scales up against a spec already known to
+// be bad. Only the reconcile after that takes the early return. The gate has to
+// be the runners themselves.
+func TestEphemeralRunnerSetIgnoresTheTargetCountOnTheReconcileThatFindsOutdatedRunners(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, clientgoscheme.AddToScheme(scheme))
+	require.NoError(t, v1alpha1.AddToScheme(scheme))
+
+	const (
+		name      = "test-ers"
+		namespace = "default"
+	)
+
+	ephemeralRunnerSet := &v1alpha1.EphemeralRunnerSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:       name,
+			Namespace:  namespace,
+			Finalizers: []string{EphemeralRunnerSetFinalizerName},
+		},
+		Spec: v1alpha1.EphemeralRunnerSetSpec{
+			// The listener has published a fresh target since the rejection.
+			// It knows nothing about it: it counts queued jobs, not whether the
+			// runners it asks for can run at all.
+			Replicas:           5,
+			PatchID:            4,
+			ActionableRevision: 2,
+			EphemeralRunnerSpec: v1alpha1.EphemeralRunnerSpec{
+				GitHubConfigURL: "https://github.com/owner/repo",
+			},
+		},
+		Status: v1alpha1.EphemeralRunnerSetStatus{
+			// Still Running: this is the reconcile that discovers the rejection.
+			Phase:                     v1alpha1.EphemeralRunnerSetPhaseRunning,
+			AppliedActionableRevision: 2,
+		},
+	}
+
+	controller := true
+	rejectedRunner := &v1alpha1.EphemeralRunner{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "runner-that-rejected-the-spec",
+			Namespace: namespace,
+			Annotations: map[string]string{
+				AnnotationKeyActionableRevision: "2",
+				AnnotationKeyPatchID:            "3",
+			},
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: v1alpha1.GroupVersion.String(),
+					Kind:       "EphemeralRunnerSet",
+					Name:       name,
+					UID:        "test-uid",
+					Controller: &controller,
+				},
+			},
+		},
+		Status: v1alpha1.EphemeralRunnerStatus{
+			Phase: v1alpha1.EphemeralRunnerPhaseOutdated,
+		},
+	}
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(ephemeralRunnerSet, rejectedRunner).
+		WithStatusSubresource(&v1alpha1.EphemeralRunnerSet{}, &v1alpha1.EphemeralRunner{}).
+		WithIndex(&v1alpha1.EphemeralRunner{}, resourceOwnerKey, newGroupVersionOwnerKindIndexer("EphemeralRunnerSet")).
+		Build()
+
+	resourceCache := NewResourceCache()
+	reconciler := &EphemeralRunnerSetReconciler{
+		Client:    fakeClient,
+		APIReader: fakeClient,
+		Log:       logr.Discard(),
+		Scheme:    scheme,
+		ResourceBuilder: ResourceBuilder{
+			ResourceCache: &resourceCache,
+			Scheme:        scheme,
+		},
+	}
+
+	key := types.NamespacedName{Namespace: namespace, Name: name}
+	_, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: key})
+	require.NoError(t, err)
+
+	runners := new(v1alpha1.EphemeralRunnerList)
+	require.NoError(t, fakeClient.List(context.Background(), runners, client.InNamespace(namespace)))
+
+	for _, runner := range runners.Items {
+		require.Equal(
+			t,
+			rejectedRunner.Name,
+			runner.Name,
+			"no runner may be created from a spec the runners have already rejected, whatever count the listener asked for",
+		)
+	}
+
+	got := new(v1alpha1.EphemeralRunnerSet)
+	require.NoError(t, fakeClient.Get(context.Background(), key, got))
+	require.Equal(
+		t,
+		v1alpha1.EphemeralRunnerSetPhaseOutdated,
+		got.Status.Phase,
+		"the rejection must be recorded on the same reconcile that found it",
+	)
+}

controllers/actions.github.com/helpers.go

@@ -1,35 +1,20 @@
 package actionsgithubcom
 
 import (
-	"strconv"
-
 	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 )
 
-func ephemeralRunnerSetNeedsOutdatedRecovery(ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet, autoscalingRunnerSet *v1alpha1.AutoscalingRunnerSet) bool {
-	if ephemeralRunnerSet == nil || autoscalingRunnerSet == nil ||
-		autoscalingRunnerSet.Generation <= autoscalingRunnerSet.Status.ObservedGeneration {
-		return false
-	}
-
-	publishedGeneration, err := strconv.ParseInt(
-		ephemeralRunnerSet.Annotations[AnnotationKeyAutoscalingRunnerSetGeneration],
-		10,
-		64,
-	)
-	if err != nil {
-		return true
-	}
-
-	return publishedGeneration < autoscalingRunnerSet.Generation
-}
-
 // ephemeralRunnerSetActionableSpecChanged reports whether the runner spec the
 // EphemeralRunnerSet is running differs from the one derived from the
 // AutoscalingRunnerSet, in a way that requires re-applying it to the runners.
 //
+// A change reported here is one of the signals that recovers a scale set from
+// the outdated phase: changing the runner spec gives the runners a different
+// input to retry. The encompassing ephemeralRunnerSetDesiredSpecChanged helper
+// also considers EphemeralRunnerMetadata changes.
+//
 // Semantic.DeepEqual is used rather than cmp.Equal or reflect.DeepEqual because
 // it treats a nil slice/map as equal to an empty one. That matters here: most
 // PodSpec collection fields carry omitempty, so a template containing an
@@ -47,6 +32,31 @@ func ephemeralRunnerSetActionableSpecChanged(current, desired *v1alpha1.Ephemera
 	return !apiequality.Semantic.DeepEqual(current.Spec.EphemeralRunnerSpec, desired.Spec.EphemeralRunnerSpec)
 }
 
+// ephemeralRunnerSetDesiredSpecChanged reports whether anything the
+// AutoscalingRunnerSet owns in the EphemeralRunnerSet spec differs from what the
+// set is running: the runner spec itself, plus the metadata stamped onto the
+// runners the set creates.
+//
+// This is the question that decides whether a scale set may leave the outdated
+// phase. The runners rejected the spec they were handed, so only a change to
+// what they would be handed next is reason to retry.
+//
+// Replicas, PatchID and ActionableRevision are deliberately excluded. They are
+// scaling bookkeeping written by the listener and by this controller, and while
+// the set is outdated they are pinned to zero, so comparing them would report
+// drift that has nothing to do with what the runners rejected.
+func ephemeralRunnerSetDesiredSpecChanged(current, desired *v1alpha1.EphemeralRunnerSet) bool {
+	if current == nil || desired == nil {
+		return current != desired
+	}
+
+	if ephemeralRunnerSetActionableSpecChanged(current, desired) {
+		return true
+	}
+
+	return !apiequality.Semantic.DeepEqual(current.Spec.EphemeralRunnerMetadata, desired.Spec.EphemeralRunnerMetadata)
+}
+
 func nextActionableRevision(current *v1alpha1.EphemeralRunnerSet) int64 {
 	if current == nil {
 		return 1

controllers/actions.github.com/indexer.go

@@ -39,6 +39,15 @@ func SetupIndexers(mgr ctrl.Manager) error {
 		return err
 	}
 
+	if err := mgr.GetFieldIndexer().IndexField(
+		context.Background(),
+		&v1alpha1.AutoscalingListener{},
+		autoscalingRunnerSetOwnerKey,
+		autoscalingListenerRunnerSetIndexer,
+	); err != nil {
+		return err
+	}
+
 	if err := mgr.GetFieldIndexer().IndexField(
 		context.Background(),
 		&v1alpha1.EphemeralRunner{},
@@ -51,6 +60,28 @@ func SetupIndexers(mgr ctrl.Manager) error {
 	return nil
 }
 
+// autoscalingListenerRunnerSetIndexer indexes a listener by the scale set its
+// spec points back at, so the listeners belonging to a scale set can be found
+// without deriving their names. The derived name is a hash over the runner group
+// and the config URL, both of which a user may edit, so a name is only ever the
+// right lookup for the listener a scale set wants *now* - never for the ones it
+// already has.
+func autoscalingListenerRunnerSetIndexer(o client.Object) []string {
+	listener, ok := o.(*v1alpha1.AutoscalingListener)
+	if !ok {
+		return nil
+	}
+
+	return []string{autoscalingRunnerSetOwnerIndexValue(
+		listener.Spec.AutoscalingRunnerSetNamespace,
+		listener.Spec.AutoscalingRunnerSetName,
+	)}
+}
+
+func autoscalingRunnerSetOwnerIndexValue(namespace, name string) string {
+	return namespace + "/" + name
+}
+
 func newGroupVersionOwnerKindIndexer(ownerKind string, otherOwnerKinds ...string) client.IndexerFunc {
 	owners := append([]string{ownerKind}, otherOwnerKinds...)
 	return func(o client.Object) []string {