fix: revoke readiness after repeated periodic validation failures (#162)

README.md

@@ -175,7 +175,10 @@ When continuous validation is enabled:
 1. The validation container runs as a native sidecar (not just an init container)
 2. After the initial validation succeeds, the container becomes ready
 3. The validation repeats at the specified interval
-4. On validation failure, the error is logged but the container continues running
+4. On validation failure, the error is logged and, once the configured
+   `failureThreshold` of consecutive failures is reached, the container is marked
+   not-ready so Kubernetes removes the pod from the Service endpoints. A later
+   successful validation restores readiness.
 5. The readiness probe reflects the validation state
 
 #### Configuration
@@ -198,8 +201,16 @@ spec:
   continuousValidation:
     enabled: true
     interval: "10m"  # Supports s, m, h units (e.g., "30s", "5m", "1h")
+    failureThreshold: 1  # Consecutive failed re-validations before the pod is
+                         # removed from Service endpoints. Default 1 (fail-closed);
+                         # 0 never revokes readiness (legacy fail-open behavior).
 ```
 
+> **Upgrade note:** `failureThreshold` defaults to `1`, so a pod whose model
+> starts failing re-validation is now removed from rotation by default. To keep
+> the previous behavior where a pod stays ready regardless of later failures, set
+> `failureThreshold: 0`. Use a higher value to tolerate transient failures.
+
 #### Requirements
 
 - Kubernetes 1.28 or later (for native sidecar support with `restartPolicy: Always`)

api/v1alpha1/modelvalidation_types.go

@@ -17,6 +17,7 @@
 import (
 	"crypto/sha256"
 	"fmt"
+	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -133,8 +134,25 @@
 	// Minimum interval is 1m to prevent excessive CPU usage.
 	// +kubebuilder:default="5m"
 	// +kubebuilder:validation:Pattern=`^([0-9]+(\.[0-9]+)?(m|h))+$`
 	// +kubebuilder:validation:XValidation:rule="self == '' || duration(self) >= duration('1m')", message="interval must be at least 1m"
 	Interval string `json:"interval,omitempty"`
+
+	// FailureThreshold is the number of consecutive failed periodic re-validations
+	// after which the pod is marked not-ready and removed from the Service
+	// endpoints, until a later periodic validation succeeds again.
+	// Only used when Enabled is true.
+	//
+	// A value of 0 disables readiness revocation (fail-open): once the pod has
+	// become ready, it stays ready regardless of later periodic failures. This
+	// matches the pre-1.0 behavior.
+	//
+	// The default is 1 (fail-closed): a single failed periodic validation removes
+	// the pod from rotation. Set a higher value to tolerate transient failures
+	// (e.g. 3 only revokes readiness after three consecutive failures).
+	// +kubebuilder:default=1
+	// +kubebuilder:validation:Minimum=0
+	// +kubebuilder:validation:Optional
+	FailureThreshold int32 `json:"failureThreshold"`
 }
 
 // ModelValidationSpec defines the desired state of ModelValidation
@@ -259,6 +277,8 @@
 		if mv.Spec.ContinuousValidation.Enabled {
 			hasher.Write([]byte("continuous-enabled"))
 			hasher.Write([]byte(mv.Spec.ContinuousValidation.Interval))
+			hasher.Write([]byte("failure-threshold:" +
+				strconv.Itoa(int(mv.Spec.ContinuousValidation.FailureThreshold))))
 		} else {
 			hasher.Write([]byte("continuous-disabled"))
 		}

cmd/validation-agent/main.go

@@ -34,17 +34,25 @@ func main() {
 	var interval time.Duration
 	var healthPort int
 	var skipInitial bool
+	var maxConsecutiveFailures int
 
 	flag.DurationVar(&interval, "interval", 0, "Validation interval (e.g., 5m, 1h). If 0 or not set, runs once and exits.")
 	flag.IntVar(&healthPort, "health-port", 8080, "Health check server port")
 	flag.BoolVar(&skipInitial, "skip-initial", false,
 		"Skip initial validation (used with legacy sidecar mode where init container already validated)")
+	flag.IntVar(&maxConsecutiveFailures, "max-consecutive-failures", 0,
+		"Number of consecutive periodic validation failures after which readiness is revoked. "+
+			"0 means never revoke readiness once ready (fail-open).")
 	flag.Parse()
 
 	log.SetLogger(zap.New())
 	logger := log.Log.WithName("validation-agent")
 
-	logger.Info("Starting validation agent", "interval", interval, "healthPort", healthPort, "skipInitial", skipInitial)
+	logger.Info("Starting validation agent",
+		"interval", interval,
+		"healthPort", healthPort,
+		"skipInitial", skipInitial,
+		"maxConsecutiveFailures", maxConsecutiveFailures)
 
 	// Setup signal handling
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
@@ -96,28 +104,17 @@ func main() {
 
 	// Continuous validation loop
 	if interval > 0 {
-		logger.Info("Starting continuous validation loop")
+		logger.Info("Starting continuous validation loop", "maxConsecutiveFailures", maxConsecutiveFailures)
 		ticker := time.NewTicker(interval)
 		defer ticker.Stop()
 
-		for {
-			select {
-			case <-ticker.C:
-				logger.Info("Running periodic validation")
-				if err := runValidation(ctx, validationArgs, logger, "periodic"); err != nil {
-					logger.Error(err, "Validation failed")
-					// Don't unmark ready - once ready, stay ready
-					// This allows the pod to continue running despite transient failures
-				} else {
-					logger.Info("Validation successful")
-					markReady() // Ensure ready state persists
-				}
-			case <-ctx.Done():
-				logger.Info("Shutting down gracefully")
-				<-healthServerDone
-				return
-			}
-		}
+		continuousValidationLoop(ctx, ticker.C, maxConsecutiveFailures, func() error {
+			return runValidation(ctx, validationArgs, logger, "periodic")
+		}, logger)
+
+		logger.Info("Shutting down gracefully")
+		<-healthServerDone
+		return
 	}
 	// One-shot mode: validation complete, exit immediately (for init containers)
 	logger.Info("One-shot validation complete, exiting")
@@ -126,6 +123,54 @@ func main() {
 	// Exit code 0 (success already handled, failures exit earlier at line 52)
 }
 
+// continuousValidationLoop runs validate on every tick until ctx is cancelled.
+// maxConsecutiveFailures controls when readiness is revoked after repeated
+// failures; see handleValidationResult for the exact semantics.
+func continuousValidationLoop(
+	ctx context.Context, tick <-chan time.Time, maxConsecutiveFailures int,
+	validate func() error, logger logr.Logger,
+) {
+	consecutiveFailures := 0
+	for {
+		select {
+		case <-tick:
+			logger.Info("Running periodic validation")
+			err := validate()
+			consecutiveFailures = handleValidationResult(err, consecutiveFailures, maxConsecutiveFailures)
+			if err != nil {
+				logger.Error(err, "Validation failed",
+					"consecutiveFailures", consecutiveFailures,
+					"maxConsecutiveFailures", maxConsecutiveFailures,
+					"ready", isReady())
+			} else {
+				logger.Info("Validation successful")
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// handleValidationResult updates readiness based on a validation outcome and
+// returns the new consecutive-failure count.
+//
+// On success the counter resets to 0 and the agent is marked ready. On failure
+// the counter increments; when maxConsecutiveFailures > 0 and the counter
+// reaches it, readiness is revoked (fail-closed). A maxConsecutiveFailures of 0
+// never revokes readiness (fail-open), preserving the legacy behavior.
+func handleValidationResult(err error, consecutiveFailures, maxConsecutiveFailures int) int {
+	if err == nil {
+		markReady()
+		return 0
+	}
+
+	consecutiveFailures++
+	if maxConsecutiveFailures > 0 && consecutiveFailures >= maxConsecutiveFailures {
+		markNotReady()
+	}
+	return consecutiveFailures
+}
+
 func runValidation(ctx context.Context, args []string, logger logr.Logger, validationType string) error {
 	tracer := otel.Tracer("validation-agent")
 	ctx, span := tracer.Start(ctx, "model.validate",
@@ -164,10 +209,7 @@ func startHealthServer(ctx context.Context, port int, logger logr.Logger) {
 	})
 
 	mux.HandleFunc("/ready", func(w http.ResponseWriter, _ *http.Request) {
-		readyMutex.RLock()
-		defer readyMutex.RUnlock()
-
-		if ready {
+		if isReady() {
 			w.WriteHeader(http.StatusOK)
 			_, _ = w.Write([]byte("ready"))
 		} else {
@@ -210,3 +252,15 @@ func markReady() {
 	defer readyMutex.Unlock()
 	ready = true
 }
+
+func markNotReady() {
+	readyMutex.Lock()
+	defer readyMutex.Unlock()
+	ready = false
+}
+
+func isReady() bool {
+	readyMutex.RLock()
+	defer readyMutex.RUnlock()
+	return ready
+}

cmd/validation-agent/main_test.go

@@ -0,0 +1,169 @@
+// Copyright 2025 The Sigstore Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// resetReady puts the shared readiness state into a known value for a test.
+// The agent uses a package-level ready flag, so tests must set it explicitly.
+func resetReady(t *testing.T, value bool) {
+	t.Helper()
+	if value {
+		markReady()
+	} else {
+		markNotReady()
+	}
+}
+
+func TestHandleValidationResult_SuccessMarksReadyAndResets(t *testing.T) {
+	resetReady(t, false)
+
+	// A success after prior failures resets the counter and marks ready.
+	count := handleValidationResult(nil, 5, 3)
+
+	assert.Equal(t, 0, count, "success must reset the consecutive-failure count")
+	assert.True(t, isReady(), "success must mark the agent ready")
+}
+
+func TestHandleValidationResult_FailOpenNeverRevokes(t *testing.T) {
+	resetReady(t, true)
+
+	// maxConsecutiveFailures == 0 means fail-open: readiness is never revoked.
+	count := 0
+	for i := 0; i < 10; i++ {
+		count = handleValidationResult(errors.New("boom"), count, 0)
+		require.True(t, isReady(), "fail-open must keep the agent ready after failure %d", i+1)
+	}
+	assert.Equal(t, 10, count, "the counter still increments while fail-open")
+}
+
+func TestHandleValidationResult_ThresholdOneRevokesImmediately(t *testing.T) {
+	resetReady(t, true)
+
+	count := handleValidationResult(errors.New("boom"), 0, 1)
+
+	assert.Equal(t, 1, count)
+	assert.False(t, isReady(), "threshold of 1 must revoke readiness on the first failure")
+}
+
+func TestHandleValidationResult_RevokesOnlyAtThreshold(t *testing.T) {
+	resetReady(t, true)
+
+	count := 0
+
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	assert.Equal(t, 1, count)
+	assert.True(t, isReady(), "must stay ready before reaching the threshold")
+
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	assert.Equal(t, 2, count)
+	assert.True(t, isReady(), "must stay ready before reaching the threshold")
+
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	assert.Equal(t, 3, count)
+	assert.False(t, isReady(), "must revoke readiness once the threshold is reached")
+}
+
+func TestHandleValidationResult_SuccessResetsBeforeThreshold(t *testing.T) {
+	resetReady(t, true)
+
+	count := 0
+	// Two failures, then a success: the counter resets so a later streak starts over.
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	require.Equal(t, 2, count)
+	require.True(t, isReady())
+
+	count = handleValidationResult(nil, count, 3)
+	require.Equal(t, 0, count)
+	require.True(t, isReady())
+
+	// Two more failures must not revoke readiness: the streak is only 2, not 3.
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	count = handleValidationResult(errors.New("boom"), count, 3)
+	assert.Equal(t, 2, count)
+	assert.True(t, isReady(), "a success must prevent non-consecutive failures from revoking readiness")
+}
+
+func TestHandleValidationResult_RecoversAfterRevoke(t *testing.T) {
+	resetReady(t, true)
+
+	// Revoke via threshold of 1, then recover on the next success.
+	count := handleValidationResult(errors.New("boom"), 0, 1)
+	require.False(t, isReady())
+
+	count = handleValidationResult(nil, count, 1)
+	assert.Equal(t, 0, count)
+	assert.True(t, isReady(), "a successful validation must restore readiness")
+}
+
+func TestContinuousValidationLoop_RevokesReadinessOnTick(t *testing.T) {
+	resetReady(t, true)
+
+	tick := make(chan time.Time)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		continuousValidationLoop(ctx, tick, 1, func() error {
+			return errors.New("boom")
+		}, logr.Discard())
+		close(done)
+	}()
+
+	tick <- time.Now()
+
+	require.Eventually(t, func() bool {
+		return !isReady()
+	}, time.Second, 5*time.Millisecond, "loop must revoke readiness after a failing tick")
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("loop did not return after context cancellation")
+	}
+}
+
+func TestContinuousValidationLoop_StopsOnContextCancel(t *testing.T) {
+	resetReady(t, true)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		continuousValidationLoop(ctx, make(chan time.Time), 1, func() error {
+			t.Error("validate must not be called when no tick fires")
+			return nil
+		}, logr.Discard())
+		close(done)
+	}()
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("loop did not return after context cancellation")
+	}
+}