fix(epp): keep metadata-less endpoints out of the p2p source maximum

The max-tracking loop counted endpoints the candidate loop later rejects
for nil metadata, so a metadata-less endpoint reporting the pool's
highest cached-token count inflated the maximum, every real endpoint
fell outside the one-block band, and the source header was silently
never set. One pass now collects usable sources and the maximum
together, which also computes each endpoint's cached tokens once.

Signed-off-by: nilig <nili.ifergan@gmail.com>

Author: nilig

pkg/epp/framework/plugins/requestcontrol/dataproducer/p2psource/producer.go

@@ -161,9 +161,26 @@ func (p *Producer) attrKey() string {
 // mildly prefer a one-request-shorter queue, and starve deeply-queued
 // sources. No-op when no candidate holds any cached block.
 func (p *Producer) Produce(ctx context.Context, request *scheduling.InferenceRequest, endpoints []scheduling.Endpoint) error {
+	// Endpoints without metadata cannot serve as sources (no address to
+	// join), so they must not pin the pool maximum either: an inflated
+	// maximum would exclude every real endpoint below.
+	type sourceMatch struct {
+		ep        scheduling.Endpoint
+		cached    int
+		blockSize int
+	}
 	maxCached := 0
+	var matches []sourceMatch
 	for _, ep := range endpoints {
-		if cached := p.cachedTokenCount(ep); cached > maxCached {
+		if ep.GetMetadata() == nil {
+			continue
+		}
+		cached, blockSize := p.cachedTokens(ep)
+		if cached == 0 {
+			continue
+		}
+		matches = append(matches, sourceMatch{ep: ep, cached: cached, blockSize: blockSize})
+		if cached > maxCached {
 			maxCached = cached
 		}
 	}
@@ -174,14 +191,13 @@ func (p *Producer) Produce(ctx context.Context, request *scheduling.InferenceReq
 		var cachedCounts []int
 		var weights []float64
 		total := 0.0
-		for _, ep := range endpoints {
-			cached, blockSize := p.cachedTokens(ep)
-			if ep.GetMetadata() == nil || cached == 0 || cached+blockSize < maxCached {
+		for _, m := range matches {
+			if m.cached+m.blockSize < maxCached {
 				continue
 			}
-			w := 1.0 / (1.0 + float64(waitingQueueSize(ep)))
-			candidates = append(candidates, ep)
-			cachedCounts = append(cachedCounts, cached)
+			w := 1.0 / (1.0 + float64(waitingQueueSize(m.ep)))
+			candidates = append(candidates, m.ep)
+			cachedCounts = append(cachedCounts, m.cached)
 			weights = append(weights, w)
 			total += w
 		}

pkg/epp/framework/plugins/requestcontrol/dataproducer/p2psource/producer_test.go

@@ -141,6 +141,25 @@ func TestProduce_MissingMatchInfo_TreatedAsZero(t *testing.T) {
 	assert.Equal(t, "10.0.0.2:8080", best.hostPort)
 }
 
+// A metadata-less endpoint must not pin the pool maximum: it cannot serve as
+// a source itself, and an inflated maximum would exclude every real endpoint
+// and silently suppress the stash.
+func TestProduce_MetadataNilMax_DoesNotSuppressStash(t *testing.T) {
+	ctx := utils.NewTestContext(t)
+	p := New("test", Config{MinCachedTokenDelta: 1})
+
+	noMD := scheduling.NewEndpoint(nil, nil, nil)
+	noMD.Put(p.prefixMatchDataKey.String(),
+		attrprefix.NewPrefixCacheMatchInfo(10, 4, testBlockSize).WithCachedBlockCount(10))
+
+	req := &scheduling.InferenceRequest{RequestID: "req-nil-md"}
+	require.NoError(t, p.Produce(ctx, req, []scheduling.Endpoint{noMD, endpoint(p, "pod-b", "10.0.0.2", 2)}))
+
+	best, ok := scheduling.ReadRequestAttribute[*bestMatchPeer](req, p.attrKey())
+	require.True(t, ok)
+	assert.Equal(t, "10.0.0.2:8080", best.hostPort)
+}
+
 // Best peer exceeds the decode pod's cached tokens by >= delta: header set.
 func TestPreRequest_SetsKVCacheSourceHeader(t *testing.T) {
 	ctx := utils.NewTestContext(t)