executor: treat a cached function service as invalid during scale-in

IsValid accepted a deployment as long as its status reported an
available replica. When the idle reaper scales the deployment to zero,
Spec.Replicas drops right away but the status can still count the pod
that is shutting down, so a cache hit kept routing requests to it.

Export the availability check from executor/util, make it look at the
spec as well, and use it in the container and newdeploy IsValid paths.

pkg/executor/executortype/container/containermgr.go

@@ -209,7 +209,7 @@ func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool
 				}
 				return false
 			}
-			if currentDeploy.Status.AvailableReplicas < 1 {
+			if !executorUtils.DeploymentAvailable(currentDeploy, 1) {
 				return false
 			}
 		}

pkg/executor/executortype/container/isvalid_test.go

@@ -59,6 +59,16 @@ func TestContainerIsValid(t *testing.T) {
 		assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef)))
 	})
 
+	t.Run("invalid while the scale-in is not reflected in the status yet", func(t *testing.T) {
+		zero := int32(0)
+		depl := deplObj("default", "fn-dep", 1)
+		depl.Generation = 2
+		depl.Spec.Replicas = &zero
+		depl.Status.ObservedGeneration = 1
+		c := newContainer(svcObj("default", "fn-svc"), depl)
+		assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef)))
+	})
+
 	t.Run("invalid when the service is missing from the cache", func(t *testing.T) {
 		c := newContainer(deplObj("default", "fn-dep", 1)) // no service
 		assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef)))

pkg/executor/executortype/newdeploy/isvalid_test.go

@@ -62,6 +62,16 @@ func TestNewdeployIsValid(t *testing.T) {
 		assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef)))
 	})
 
+	t.Run("invalid while the scale-in is not reflected in the status yet", func(t *testing.T) {
+		zero := int32(0)
+		depl := deplObj("default", "fn-dep", 1)
+		depl.Generation = 2
+		depl.Spec.Replicas = &zero
+		depl.Status.ObservedGeneration = 1
+		d := newDeploy(svcObj("default", "fn-svc"), depl)
+		assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef)))
+	})
+
 	t.Run("invalid when the service is missing from the cache", func(t *testing.T) {
 		d := newDeploy(deplObj("default", "fn-dep", 1)) // no service
 		assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef)))

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -232,7 +232,7 @@ func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) boo
 				}
 				return false
 			}
-			if currentDeploy.Status.AvailableReplicas < 1 {
+			if !executorUtils.DeploymentAvailable(currentDeploy, 1) {
 				return false
 			}
 		}

pkg/executor/util/deployment.go

@@ -62,7 +62,7 @@ func nextWaitPollDelay(cur time.Duration) time.Duration {
 	return min(cur*3/2, waitPollMaxDelay)
 }
 
-// deploymentAvailable reports whether depl has at least replicas available,
+// DeploymentAvailable reports whether depl has at least replicas available,
 // as observed by the deployment controller for the spec currently stored.
 //
 // The generation gate matters on the scale-from-zero path. UpdateScale only
@@ -73,8 +73,19 @@ func nextWaitPollDelay(cur time.Duration) time.Duration {
 // report a serving deployment while the Service has no endpoint. Until
 // ObservedGeneration catches up, the status says nothing about this scale
 // request. Same gate as WaitForDeploymentRollout in the integration framework.
-func deploymentAvailable(depl *appsv1.Deployment, replicas int32) bool {
-	return depl.Status.ObservedGeneration >= depl.Generation && depl.Status.AvailableReplicas >= replicas
+//
+// The spec check covers the way down. Once the idle reaper scales the
+// deployment in, Spec.Replicas drops at once while the status can still count
+// the pod that is going away, which would keep a cached function service
+// looking valid.
+func DeploymentAvailable(depl *appsv1.Deployment, replicas int32) bool {
+	specReplicas := int32(1)
+	if depl.Spec.Replicas != nil {
+		specReplicas = *depl.Spec.Replicas
+	}
+	return specReplicas >= replicas &&
+		depl.Status.ObservedGeneration >= depl.Generation &&
+		depl.Status.AvailableReplicas >= replicas
 }
 
 // WaitForDeployment polls the deployment until the deployment controller has
@@ -108,7 +119,7 @@ func WaitForDeployment(ctx context.Context, kubeClient kubernetes.Interface, log
 			return nil, err
 		}
 		// TODO check for imagePullerror
-		if deploymentAvailable(latestDepl, replicas) {
+		if DeploymentAvailable(latestDepl, replicas) {
 			otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...)
 			return latestDepl, err
 		}