fix(flowcontrol): map pre-admission TTL/cancel to eviction statuses i…

pkg/epp/flowcontrol/controller/controller.go

@@ -355,7 +355,13 @@ 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)
+		// An internal invariant violation, not a capacity condition: finalize as RejectedOther so it
+		// surfaces as an internal error rather than as saturation backpressure. The registry error is
+		// flattened with %v because this finalized error is returned through the connection closure in
+		// EnqueueAndWait: a %w-preserved ErrPriorityBandNotFound would be misread by
+		// withConnectionWithFallback as a lease-acquisition failure and silently retried at priority 0.
+		item.FinalizeWithOutcome(types.QueueOutcomeRejectedOther,
+			fmt.Errorf("%w: failed to get ManagedQueue for leased flow: %v", types.ErrRejected, err))
 		return item, err
 	}
 

pkg/epp/flowcontrol/controller/controller_test.go

@@ -462,16 +462,21 @@ func TestFlowController_EnqueueAndWait(t *testing.T) {
 			h := newUnitHarness(t.Context(), t, &Config{}, mockRegistry, nil)
 
 			// Create a faulty setup that successfully leases the flow but fails to return the
-			// ManagedQueue. This setup should be considered as unavailable.
+			// ManagedQueue. The error wraps ErrPriorityBandNotFound (the realistic registry failure) to
+			// prove the sentinel does not leak into the finalized error, where
+			// withConnectionWithFallback would misread it as a lease-acquisition failure and retry at
+			// priority 0.
 			faultyRegistry := &mocks.MockRegistryDataPlane{
 				ManagedQueueFunc: func(_ flowcontrol.FlowKey) (contracts.ManagedQueue, error) {
-					return nil, errors.New("invariant violation: queue retrieval failed")
+					return nil, fmt.Errorf("invariant violation: %w", contracts.ErrPriorityBandNotFound)
 				},
 			}
+			var connectionAttempts int
 			mockRegistry.WithConnectionFunc = func(
 				key flowcontrol.FlowKey,
 				fn func(conn contracts.ActiveFlowConnection) error,
 			) error {
+				connectionAttempts++
 				return fn(&mockActiveFlowConnection{
 					RegistryV: faultyRegistry,
 					FlowKeyV:  key,
@@ -482,8 +487,12 @@ 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.NotErrorIs(t, err, contracts.ErrPriorityBandNotFound,
+				"registry sentinels must not leak into the finalized error")
+			assert.Equal(t, types.QueueOutcomeRejectedOther, outcome,
+				"outcome should be QueueOutcomeRejectedOther when queue doesn't exist for the flow")
+			assert.Equal(t, 1, connectionAttempts,
+				"an invariant violation on a leased flow must not trigger the priority-0 fallback retry")
 		})
 	})
 

pkg/epp/requestcontrol/admission.go

@@ -242,10 +242,20 @@ func translateFlowControlOutcome(outcome types.QueueOutcome, err error) error {
 	case types.QueueOutcomeEvictedContextCancelled:
 		return errcommon.Error{Code: errcommon.ServiceUnavailable, Msg: "client disconnected: " + msg, Headers: map[string]string{errcommon.RequestDroppedReasonHeaderKey: string(errcommon.RequestDroppedReasonContextCancelled)}}
 	case types.QueueOutcomeRejectedOther, types.QueueOutcomeEvictedOther:
-		if errors.Is(err, types.ErrFlowControllerNotRunning) {
+		switch {
+		case 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 mapping of the post-admission equivalent; the two paths then agree by construction.
+		case errors.Is(err, types.ErrTTLExpired):
+			return translateFlowControlOutcome(types.QueueOutcomeEvictedTTL, err)
+		case errors.Is(err, types.ErrContextCancelled):
+			return translateFlowControlOutcome(types.QueueOutcomeEvictedContextCancelled, err)
+		default:
+			return errcommon.Error{Code: errcommon.Internal, Msg: "internal flow control error: " + msg}
 		}
-		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,