fix(scheduling): return typed capacity rejection when filters drain candidates

A filter draining a non-empty candidate set produced an Internal error that
the scheduler dropped after logging, so profile handlers reported the failure
with fresh untyped errors and the director fell through to its legacy
ResourceExhausted fallback: the client got a 429 chosen by accident, with no
x-llm-d-request-dropped-reason header.

Report the drain with the same vocabulary as a flow control capacity
rejection (ResourceExhausted plus rejected-saturated), and retain per-profile
errors in Schedule, joining them into the ProcessResults error so the typed
code survives to the director's errors.As without touching any profile
handler.

The director's own pre-scheduling rejections (no endpoint candidates located,
screeners eliminating every candidate) carry rejected-no-endpoints on their
ServiceUnavailable responses, so every pre-dispatch rejection now reports a
machine-readable drop reason.

Fixes llm-d#2428

Signed-off-by: Luke Van Drie <lukevandrie@google.com>

Author: LukeAVanDrie

pkg/common/error/error.go

@@ -26,7 +26,7 @@ import (
 )
 
 // RequestDroppedReasonHeaderKey is the HTTP response header that communicates the specific
-// reason a request was dropped by flow control.
+// reason the EPP dropped a request.
 const RequestDroppedReasonHeaderKey = "x-llm-d-request-dropped-reason"
 
 // RequestDroppedReason is the reason a request was rejected before dispatch or evicted after dispatch.

pkg/epp/requestcontrol/director.go

@@ -304,17 +304,19 @@ func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestCo
 	endpointCandidates := d.endpointCandidates.Locate(ctx, reqCtx.Request.Metadata)
 	if len(endpointCandidates) == 0 {
 		return reqCtx, errcommon.Error{
-			Code: errcommon.ServiceUnavailable,
-			Msg:  "failed to find endpoint candidates for serving the request",
+			Code:    errcommon.ServiceUnavailable,
+			Msg:     "failed to find endpoint candidates for serving the request",
+			Headers: map[string]string{errcommon.RequestDroppedReasonHeaderKey: string(errcommon.RequestDroppedReasonNoEndpoints)},
 		}
 	}
 
 	snapshotOfCandidatePods := d.toSchedulerEndpoints(endpointCandidates)
 	snapshotOfCandidatePods = d.runScreeners(ctx, reqCtx.SchedulingRequest, snapshotOfCandidatePods)
 	if len(snapshotOfCandidatePods) == 0 {
 		return reqCtx, errcommon.Error{
-			Code: errcommon.ServiceUnavailable,
-			Msg:  "screeners eliminated all endpoint candidates",
+			Code:    errcommon.ServiceUnavailable,
+			Msg:     "screeners eliminated all endpoint candidates",
+			Headers: map[string]string{errcommon.RequestDroppedReasonHeaderKey: string(errcommon.RequestDroppedReasonNoEndpoints)},
 		}
 	}
 	// Prepare per request data by running DataProducer plugins.

pkg/epp/requestcontrol/director_test.go

@@ -414,11 +414,13 @@ func TestDirector_HandleRequest(t *testing.T) {
 		initialTargetModelName  string // Initial target model in the reqCtx.
 		parser                  fwkrh.Parser
 		wantErrCode             string                   // Expected errcommon code string
+		wantDroppedReason       string                   // If non-empty, expected x-llm-d-request-dropped-reason header on the error
 		wantReqCtx              *handlers.RequestContext // Fields to check in the returned RequestContext
 		targetModelName         string                   // Expected model name after target model resolution
 		admitRequestDenialError error                    // Expected denial error from admission plugin
 		dataProducerPlugin      *mockDataProducerPlugin
 		screener                *mockScreener
+		emptyEndpoints          bool // If true, the director locates no endpoint candidates.
 		preRequestPlugins       []*mockPreRequestPlugin
 		requestHeaderPlugin     *mockRequestHeaderPlugin
 		wantMutatedBody         map[string]any
@@ -995,7 +997,21 @@ func TestDirector_HandleRequest(t *testing.T) {
 			screener: &mockScreener{name: "eliminate-all", screen: func([]fwksched.Endpoint) []fwksched.Endpoint {
 				return nil
 			}},
-			wantErrCode: errcommon.ServiceUnavailable,
+			wantErrCode:       errcommon.ServiceUnavailable,
+			wantDroppedReason: string(errcommon.RequestDroppedReasonNoEndpoints),
+		},
+		{
+			name: "no endpoint candidates located",
+			reqBodyMap: map[string]any{
+				"model":  model,
+				"prompt": "critical prompt",
+			},
+			mockAdmissionController: &mockAdmissionController{admitErr: nil},
+			initialTargetModelName:  model,
+			inferenceObjectiveName:  objectiveName,
+			emptyEndpoints:          true,
+			wantErrCode:             errcommon.ServiceUnavailable,
+			wantDroppedReason:       string(errcommon.RequestDroppedReasonNoEndpoints),
 		},
 		{
 			name: "scheduler returns error",
@@ -1010,6 +1026,30 @@ func TestDirector_HandleRequest(t *testing.T) {
 			wantErrCode:            errcommon.ResourceExhausted,
 			inferenceObjectiveName: objectiveName,
 		},
+		{
+			// The typed error inside a joined scheduler error, including its
+			// drop-reason header, must reach the caller instead of the
+			// untyped-error fallback.
+			name: "scheduler returns joined error with typed capacity rejection",
+			reqBodyMap: map[string]any{
+				"model":  model,
+				"prompt": "prompt that causes scheduling drain",
+			},
+			mockAdmissionController: &mockAdmissionController{admitErr: nil},
+			schedulerMockSetup: func(m *mockScheduler) {
+				m.scheduleErr = errors.Join(
+					errors.New("failed to run scheduler profile 'default'"),
+					fmt.Errorf("profile %q: %w", "default", errcommon.Error{
+						Code:    errcommon.ResourceExhausted,
+						Msg:     "no endpoints available for the given request",
+						Headers: map[string]string{errcommon.RequestDroppedReasonHeaderKey: string(errcommon.RequestDroppedReasonSaturated)},
+					}),
+				)
+			},
+			wantErrCode:            errcommon.ResourceExhausted,
+			wantDroppedReason:      string(errcommon.RequestDroppedReasonSaturated),
+			inferenceObjectiveName: objectiveName,
+		},
 		{
 			name: "scheduler returns nil result and nil error",
 			reqBodyMap: map[string]any{
@@ -1090,6 +1130,9 @@ func TestDirector_HandleRequest(t *testing.T) {
 
 				endpointCandidates := NewCachedEndpointCandidates(context.Background(), NewDatastoreEndpointCandidates(ds), time.Minute)
 				director := NewDirectorWithConfig(ds, mockSched, test.mockAdmissionController, endpointCandidates, config)
+				if test.emptyEndpoints {
+					director.endpointCandidates = &mockEndpointCandidates{}
+				}
 				if len(test.rewrites) > 0 {
 					mockDs := &mockDatastore{
 						pods:     ds.PodList(datastore.AllPodsPredicate),
@@ -1140,6 +1183,9 @@ func TestDirector_HandleRequest(t *testing.T) {
 					var e errcommon.Error
 					if assert.ErrorAs(t, err, &e, "Error should be of type errcommon.Error") {
 						assert.Equal(t, test.wantErrCode, e.Code, "Error code mismatch")
+						if test.wantDroppedReason != "" {
+							assert.Equal(t, test.wantDroppedReason, e.Headers[errcommon.RequestDroppedReasonHeaderKey], "drop-reason header mismatch")
+						}
 					}
 					return
 				}

pkg/epp/scheduling/scheduler.go

@@ -19,7 +19,10 @@ package scheduling
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"maps"
+	"slices"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -64,6 +67,11 @@ func (s *Scheduler) Schedule(ctx context.Context, request *fwksched.InferenceReq
 	}()
 
 	profileRunResults := map[string]*fwksched.ProfileRunResult{}
+	// Keyed like profileRunResults so a profile that fails in one iteration and
+	// succeeds in a later one leaves no stale error behind. Nil until a profile
+	// fails: delete and len are no-ops on a nil map, and the happy path skips
+	// the allocation.
+	var profileRunErrors map[string]error
 
 	for { // get the next set of profiles to run iteratively based on the request and the previous execution results
 		loggerVerbose.Info("Running profile handler, Pick profiles", "plugin", s.profileHandler.TypedName())
@@ -81,8 +89,13 @@ func (s *Scheduler) Schedule(ctx context.Context, request *fwksched.InferenceReq
 			profileRunResult, err := runSchedulerProfile(ctx, name, profile, request, candidateEndpoints)
 			if err != nil {
 				loggerVerbose.Info("failed to run scheduler profile", "profile", name, "error", err.Error())
+				if profileRunErrors == nil {
+					profileRunErrors = map[string]error{}
+				}
+				profileRunErrors[name] = fmt.Errorf("profile %q: %w", name, err)
 			} else {
 				loggerVerbose.Info("Completed running scheduler profile succuessfully", "profile", name)
+				delete(profileRunErrors, name)
 			}
 
 			profileRunResults[name] = profileRunResult // if profile failed to run, the run result is nil
@@ -100,6 +113,21 @@ func (s *Scheduler) Schedule(ctx context.Context, request *fwksched.InferenceReq
 	metrics.RecordPluginProcessingLatency(processProfilesResultsExtensionPoint, s.profileHandler.TypedName().Type, s.profileHandler.TypedName().Name, time.Since(before))
 	loggerVerbose.Info("Completed running profile handler ProcessResults successfully", "plugin", s.profileHandler.TypedName())
 
+	// Profile handlers see failed profiles only as nil results and report them
+	// with fresh untyped errors. Join the retained profile errors so a typed
+	// errcommon.Error raised inside a profile run (e.g. filters draining the
+	// candidate set) stays reachable via errors.As in the caller.
+	if err != nil && len(profileRunErrors) > 0 {
+		errs := make([]error, 0, len(profileRunErrors)+1)
+		errs = append(errs, err)
+		// Sorted so error composition, and therefore errors.As selection when
+		// profiles fail with different typed codes, is deterministic.
+		for _, name := range slices.Sorted(maps.Keys(profileRunErrors)) {
+			errs = append(errs, profileRunErrors[name])
+		}
+		err = errors.Join(errs...)
+	}
+
 	return result, err
 }
 

pkg/epp/scheduling/scheduler_profile.go

@@ -126,7 +126,15 @@ func (p *SchedulerProfile) String() string {
 func (p *SchedulerProfile) Run(ctx context.Context, request *fwksched.InferenceRequest, candidateEndpoints []fwksched.Endpoint) (*fwksched.ProfileRunResult, error) {
 	endpoints := p.runFilterPlugins(ctx, request, candidateEndpoints)
 	if len(endpoints) == 0 {
-		return nil, errcommon.Error{Code: errcommon.Internal, Msg: "no endpoints available for the given request"}
+		// Filters draining a non-empty candidate set means the pool is busy, not
+		// broken: an empty pool is rejected in the director before scheduling
+		// runs. Report it with the same status and drop-reason vocabulary as a
+		// flow control capacity rejection.
+		return nil, errcommon.Error{
+			Code:    errcommon.ResourceExhausted,
+			Msg:     "no endpoints available for the given request",
+			Headers: map[string]string{errcommon.RequestDroppedReasonHeaderKey: string(errcommon.RequestDroppedReasonSaturated)},
+		}
 	}
 	// if we got here, there is at least one endpoint to score
 	weightedScorePerEndpoint := p.runScorerPlugins(ctx, request, endpoints)

pkg/epp/scheduling/scheduler_test.go

@@ -18,6 +18,7 @@ package scheduling
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -26,7 +27,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 
+	errcommon "github.com/llm-d/llm-d-router/pkg/common/error"
 	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
+	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
 	fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/picker"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/picker/maxscore"
@@ -159,3 +162,35 @@ func TestSchedule(t *testing.T) {
 		})
 	}
 }
+
+// Tests that a filter draining the candidate set surfaces a typed capacity
+// rejection from Schedule.
+func TestScheduleFilterDrainReturnsTypedError(t *testing.T) {
+	drainingFilter := &testPlugin{typedName: fwkplugin.TypedName{Type: "drain-filter", Name: "drain-filter"}} // empty FilterRes drops every endpoint
+
+	profile := NewSchedulerProfile().
+		WithFilters(drainingFilter).
+		WithPicker(maxscore.NewMaxScorePicker(picker.DefaultMaxNumOfEndpoints))
+
+	schedulerConfig := NewSchedulerConfig(single.NewSingleProfileHandler(), map[string]fwksched.SchedulerProfile{"default": profile})
+	scheduler := NewSchedulerWithConfig(schedulerConfig)
+
+	req := &fwksched.InferenceRequest{
+		RequestID:   uuid.NewString(),
+		TargetModel: "any-model",
+	}
+	input := []fwksched.Endpoint{
+		fwksched.NewEndpoint(&fwkdl.EndpointMetadata{ID: k8stypes.NamespacedName{Name: "pod1"}}, &fwkdl.Metrics{}, nil),
+	}
+
+	result, err := scheduler.Schedule(context.Background(), req, input)
+	assert.Nil(t, result)
+	assert.Error(t, err)
+
+	var typedErr errcommon.Error
+	if !errors.As(err, &typedErr) {
+		t.Fatalf("Schedule error is not an errcommon.Error: %v", err)
+	}
+	assert.Equal(t, errcommon.ResourceExhausted, typedErr.Code)
+	assert.Equal(t, string(errcommon.RequestDroppedReasonSaturated), typedErr.Headers[errcommon.RequestDroppedReasonHeaderKey])
+}