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

A TTL expiry or client disconnect before an item reaches a queue
(buffered in the enqueue channel, blocked in SubmitOrBlock, or racing
managedQueue.Add) finalizes as QueueOutcomeRejectedOther, which the
translator mapped to 500. Once the item is queued, the same event maps
to 503 with a dropped-reason header. These pre-admission windows are
widest under overload, so a saturation stall with TTL expiries showed
up as a spike of 500s that was really backpressure.

The translator now recognizes ErrTTLExpired and ErrContextCancelled in
the RejectedOther/EvictedOther arm (after the shutdown check, which
keeps precedence) and delegates to the matching eviction mapping, so
the two paths cannot drift apart. The same branches cover EvictedOther
for symmetry; nothing currently emits it with these sentinels.

Separately, tryDistribution finalized a ManagedQueue lookup failure for
a leased flow as QueueOutcomeRejectedCapacity, reporting an internal
invariant violation as 429 with the Saturated header. It now finalizes
as RejectedOther: a 500 with no saturation header. This also keeps the
RejectedCapacity metric label to genuine capacity rejections. The
registry error is flattened with %v because wrapping it would let
ErrPriorityBandNotFound flow back through the connection closure and
trip the priority-0 fallback in withConnectionWithFallback.

Fixes #2097
Part of #1187

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

Author: LukeAVanDrie

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

@@ -397,16 +397,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,
@@ -417,8 +422,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,