diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go index 88bff4a0be..efbe7ad6a1 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/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 } diff --git a/pkg/epp/flowcontrol/controller/controller_test.go b/pkg/epp/flowcontrol/controller/controller_test.go index 6ca46d88f1..8c2cc14d0e 100644 --- a/pkg/epp/flowcontrol/controller/controller_test.go +++ b/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") }) }) diff --git a/pkg/epp/requestcontrol/admission.go b/pkg/epp/requestcontrol/admission.go index 8543a071b5..65229f0d2e 100644 --- a/pkg/epp/requestcontrol/admission.go +++ b/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} } diff --git a/pkg/epp/requestcontrol/admission_test.go b/pkg/epp/requestcontrol/admission_test.go index ecf7f1e5d1..43cf55a067 100644 --- a/pkg/epp/requestcontrol/admission_test.go +++ b/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,