fix(flowcontrol): map pre-admission TTL/cancel to eviction statuses instead of 500

A TTL expiry or client disconnect that fires before an item is admitted
to a queue (buffered in the enqueue channel, blocked in SubmitOrBlock,
or racing managedQueue.Add) surfaces as QueueOutcomeRejectedOther
wrapping ErrTTLExpired or ErrContextCancelled, which
translateFlowControlOutcome mapped to 500 Internal. The same client
behavior an instant later (item queued) yields EvictedTTL or
EvictedContextCancelled and a 503 with a dropped-reason header.
translateFlowControlOutcome now branches on those sentinels in the
RejectedOther/EvictedOther case (after the shutdown check, which takes
precedence) and delegates to the corresponding eviction outcome's
mapping, so the pre-admission path always follows the post-admission
one, including any future changes to that mapping. The EvictedOther arm
gets the same branches as defensive symmetry; no current producer emits
it with these sentinels.

tryDistribution finalized a ManagedQueue lookup failure for a leased
flow (an internal invariant violation) as QueueOutcomeRejectedCapacity,
surfacing an internal error as 429 with the Saturated dropped-reason
header. It now finalizes as QueueOutcomeRejectedOther with a
descriptive error, which maps to 500 Internal without a saturation
header.

Fixes llm-d#2097
Part of llm-d#1187

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

Author: LukeAVanDrie

pkg/epp/flowcontrol/controller/controller.go

@@ -355,7 +355,10 @@ func (fc *FlowController) tryDistribution(
 		fc.logger.Error(err,
 			"Invariant violation. Failed to get ManagedQueue for a leased flow.",
 			"flowKey", conn.FlowKey())
-		item.FinalizeWithOutcome(types.QueueOutcomeRejectedCapacity, types.ErrRejected)
+		// This is an internal invariant violation, not a capacity condition: finalize as RejectedOther so
+		// it surfaces as an internal error rather than as saturation backpressure.
+		item.FinalizeWithOutcome(types.QueueOutcomeRejectedOther,
+			fmt.Errorf("%w: failed to get ManagedQueue for leased flow: %w", types.ErrRejected, err))
 		return item, err
 	}
 

pkg/epp/flowcontrol/controller/controller_test.go

@@ -417,8 +417,8 @@ func TestFlowController_EnqueueAndWait(t *testing.T) {
 			outcome, err := h.fc.EnqueueAndWait(context.Background(), req)
 			require.Error(t, err, "EnqueueAndWait must reject requests if queue doesn't exist for flow")
 			assert.ErrorIs(t, err, types.ErrRejected, "error should wrap ErrRejected")
-			assert.Equal(t, types.QueueOutcomeRejectedCapacity, outcome,
-				"outcome should be QueueOutcomeRejectedCapacity when queue doesn't exist for the flow")
+			assert.Equal(t, types.QueueOutcomeRejectedOther, outcome,
+				"outcome should be QueueOutcomeRejectedOther when queue doesn't exist for the flow")
 		})
 	})
 

pkg/epp/requestcontrol/admission.go

@@ -245,6 +245,16 @@ func translateFlowControlOutcome(outcome types.QueueOutcome, err error) error {
 		if errors.Is(err, types.ErrFlowControllerNotRunning) {
 			return errcommon.Error{Code: errcommon.ServiceUnavailable, Msg: "flow controller shutting down: " + msg, Headers: map[string]string{errcommon.RequestDroppedReasonHeaderKey: string(errcommon.RequestDroppedReasonShuttingDown)}}
 		}
+		// A TTL expiry or client disconnect that fires before the item is admitted to a queue (e.g. while
+		// buffered in the enqueue channel or blocked in submission) surfaces as RejectedOther/EvictedOther
+		// rather than as a dedicated eviction outcome. These are client-caused terminations, so delegate
+		// to the same mapping as their post-admission equivalents.
+		if errors.Is(err, types.ErrTTLExpired) {
+			return translateFlowControlOutcome(types.QueueOutcomeEvictedTTL, err)
+		}
+		if errors.Is(err, types.ErrContextCancelled) {
+			return translateFlowControlOutcome(types.QueueOutcomeEvictedContextCancelled, err)
+		}
 		return errcommon.Error{Code: errcommon.Internal, Msg: "internal flow control error: " + msg}
 	default:
 		return errcommon.Error{Code: errcommon.Internal, Msg: "unhandled flow control outcome: " + msg}

pkg/epp/requestcontrol/admission_test.go

@@ -19,6 +19,7 @@ package requestcontrol
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -364,6 +365,41 @@ func TestTranslateFlowControlOutcome(t *testing.T) {
 			wantCode:   errcommon.ServiceUnavailable,
 			wantReason: string(errcommon.RequestDroppedReasonShuttingDown),
 		},
+		{
+			name:       "pre-admission TTL rejection maps like TTL eviction",
+			outcome:    fctypes.QueueOutcomeRejectedOther,
+			err:        fmt.Errorf("%w: %w", fctypes.ErrRejected, fctypes.ErrTTLExpired),
+			wantCode:   errcommon.ServiceUnavailable,
+			wantReason: string(errcommon.RequestDroppedReasonTTLExpired),
+		},
+		{
+			name:       "pre-admission cancellation rejection maps like cancellation eviction",
+			outcome:    fctypes.QueueOutcomeRejectedOther,
+			err:        fmt.Errorf("%w: %w", fctypes.ErrRejected, fctypes.ErrContextCancelled),
+			wantCode:   errcommon.ServiceUnavailable,
+			wantReason: string(errcommon.RequestDroppedReasonContextCancelled),
+		},
+		{
+			name:       "other TTL eviction maps like TTL eviction",
+			outcome:    fctypes.QueueOutcomeEvictedOther,
+			err:        fmt.Errorf("%w: %w", fctypes.ErrEvicted, fctypes.ErrTTLExpired),
+			wantCode:   errcommon.ServiceUnavailable,
+			wantReason: string(errcommon.RequestDroppedReasonTTLExpired),
+		},
+		{
+			name:       "other cancellation eviction maps like cancellation eviction",
+			outcome:    fctypes.QueueOutcomeEvictedOther,
+			err:        fmt.Errorf("%w: %w", fctypes.ErrEvicted, fctypes.ErrContextCancelled),
+			wantCode:   errcommon.ServiceUnavailable,
+			wantReason: string(errcommon.RequestDroppedReasonContextCancelled),
+		},
+		{
+			name:       "shutdown takes precedence over TTL",
+			outcome:    fctypes.QueueOutcomeRejectedOther,
+			err:        fmt.Errorf("%w: %w", fctypes.ErrFlowControllerNotRunning, fctypes.ErrTTLExpired),
+			wantCode:   errcommon.ServiceUnavailable,
+			wantReason: string(errcommon.RequestDroppedReasonShuttingDown),
+		},
 		{
 			name:     "internal error returns 500",
 			outcome:  fctypes.QueueOutcomeRejectedOther,