fix(epp): clamp encoder matched size to input size

The predictor rejects matched > input, so inconsistent match data or
mismatched feature slices would fail validation and drop the whole
prediction or training entry instead of degrading gracefully.

Signed-off-by: Kaushik Mitra <kaushikmitra@google.com>

Author: kaushikmitr

pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks.go

@@ -136,6 +136,15 @@ func (pl *PredictedLatency) captureEncoderCacheSizes(ctx context.Context, predic
 	}
 	inputSize := sumItemSizes(matchInfo.RequestItems())
 	matchedSize := sumItemSizes(matchInfo.MatchedItems())
+	// The predictor rejects matched > input, so clamp inconsistent match data
+	// rather than dropping the whole prediction or training entry.
+	if matchedSize > inputSize {
+		logger.V(logutil.DEBUG).Info("Encoder cache matched size exceeds input size, clamping",
+			"pod", endpoint.GetMetadata().String(),
+			"encoderInputSize", inputSize,
+			"encoderMatchedSize", matchedSize)
+		matchedSize = inputSize
+	}
 	predictedLatencyCtx.encoderInputSize = inputSize
 	predictedLatencyCtx.encoderMatchedSizeForEndpoints[endpoint.GetMetadata().NamespacedName.Name] = matchedSize
 	logger.V(logutil.DEBUG).Info("Encoder cache sizes for pod",

pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks_test.go

@@ -78,6 +78,30 @@ func TestProduce_CapturesEncoderCacheSizes(t *testing.T) {
 	assert.Equal(t, 0, plCtx.encoderMatchedSizeForEndpoints["pod-unmatched"])
 }
 
+// TestProduce_ClampsInconsistentEncoderMatchData verifies that match data
+// whose matched size exceeds the input size is clamped so downstream
+// predictor validation (matched <= input) cannot reject the request.
+func TestProduce_ClampsInconsistentEncoderMatchData(t *testing.T) {
+	cfg := DefaultConfig
+	cfg.PredictInProduce = false
+	cfg.UseEncoderCacheFeatures = true
+	pl := NewPredictedLatency(LatencyDataProviderPluginType, cfg, nil)
+
+	request := createTestInferenceRequest("encoder-clamp-test", 0, 0)
+	endpoint := createTestEndpoint("pod-a", 0.1, 0, 0)
+
+	matched := []attrmm.MatchItem{{Hash: "img-a", Size: 3}}
+	requestItems := []attrmm.MatchItem{{Hash: "img-b", Size: 1}}
+	endpoint.Put(pl.encoderCacheDataKey.String(), attrmm.NewEncoderCacheMatchInfo(matched, requestItems))
+
+	require.NoError(t, pl.Produce(context.Background(), request, []fwksched.Endpoint{endpoint}))
+
+	plCtx, err := pl.getPredictedLatencyContextForRequest(request)
+	require.NoError(t, err)
+	assert.Equal(t, 1, plCtx.encoderInputSize)
+	assert.Equal(t, 1, plCtx.encoderMatchedSizeForEndpoints["pod-a"])
+}
+
 // TestProduce_EncoderCacheFeatureDisabledIgnoresMatchData is the negative
 // control: with the feature off, attached match data is not read.
 func TestProduce_EncoderCacheFeatureDisabledIgnoresMatchData(t *testing.T) {

pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training.go

@@ -227,7 +227,9 @@ func bulkPredictWithMetrics(
 			bulkRequests[i].EncoderInputSize = encoderInputSizes[i]
 		}
 		if i < len(encoderMatchedSizes) {
-			bulkRequests[i].EncoderMatchedSize = encoderMatchedSizes[i]
+			// The predictor rejects matched > input, so clamp instead of
+			// failing the whole bulk request when the slices are inconsistent.
+			bulkRequests[i].EncoderMatchedSize = min(encoderMatchedSizes[i], bulkRequests[i].EncoderInputSize)
 		}
 	}
 

pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training_test.go

@@ -142,6 +142,31 @@ func TestBulkPredictWithMetrics_PropagatesEncoderSizes(t *testing.T) {
 	assert.Equal(t, 0, mockPredictor.capturedBulkStrictRequests[1].EncoderMatchedSize)
 }
 
+// TestBulkPredictWithMetrics_ClampsMatchedWithoutInputSizes verifies that a
+// matched-size slice without a corresponding input-size slice is clamped to
+// the input size (0) instead of producing requests that fail validation.
+func TestBulkPredictWithMetrics_ClampsMatchedWithoutInputSizes(t *testing.T) {
+	mockPredictor := &mockPredictor{
+		predictions: map[string]*latencypredictor.PredictionResponse{
+			"0.5": {TTFT: 0.5, TPOT: 0.03},
+		},
+	}
+
+	metricsStates := []*fwkdl.Metrics{{KVCacheUsagePercent: 0.5}}
+	pods := []*fwkdl.EndpointMetadata{
+		{NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod1"}},
+	}
+
+	_, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor,
+		metricsStates, "", pods, []int{1}, []int{1}, []float64{0.0},
+		nil, nil, nil, []int{5})
+	require.NoError(t, err)
+
+	require.Len(t, mockPredictor.capturedBulkStrictRequests, 1)
+	assert.Equal(t, 0, mockPredictor.capturedBulkStrictRequests[0].EncoderInputSize)
+	assert.Equal(t, 0, mockPredictor.capturedBulkStrictRequests[0].EncoderMatchedSize)
+}
+
 func TestBuildPredictionRequestAndTrainingEntry_EncoderSizes(t *testing.T) {
 	m := &fwkdl.Metrics{KVCacheUsagePercent: 0.5}
 	pod := &fwkdl.EndpointMetadata{NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod1"}}