rename Aborter→Evictor, Plugin→RequestEvictor,

use 429 status code, add ImmediateResponseEvictor.Cleanup()

Signed-off-by: RishabhSaini <rishabhsaini01@gmail.com>

Author: RishabhSaini

pkg/epp/flowcontrol/eviction/aborter.go

@@ -27,49 +27,54 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
 )
 
-// Aborter handles aborting an in-flight request on a model server.
-type Aborter interface {
-	Abort(ctx context.Context, item *flowcontrol.EvictionItem) error
+// Evictor handles evicting an in-flight request on a model server.
+type Evictor interface {
+	Evict(ctx context.Context, item *flowcontrol.EvictionItem) error
 }
 
-// NoOpAborter logs the eviction but does not abort the request on the model server.
-type NoOpAborter struct{}
+// NoOpEvictor logs the eviction but does not evict the request on the model server.
+type NoOpEvictor struct{}
 
-func (a *NoOpAborter) Abort(ctx context.Context, item *flowcontrol.EvictionItem) error {
-	log.FromContext(ctx).V(logutil.DEBUG).Info("Eviction selected request for abort (no-op: abort mechanism not available)",
+func (e *NoOpEvictor) Evict(ctx context.Context, item *flowcontrol.EvictionItem) error {
+	log.FromContext(ctx).V(logutil.DEBUG).Info("Eviction selected request (no-op: eviction mechanism not available)",
 		"requestID", item.RequestID,
 		"priority", item.Priority,
 		"targetURL", item.TargetURL)
 	return nil
 }
 
-// ImmediateResponseAborter aborts requests by closing the EvictionItem's AbortCh.
+// ImmediateResponseEvictor evicts requests by closing the EvictionItem's AbortCh.
 // The ext_proc Process() goroutine selects on this channel and sends an ImmediateResponse
 // to Envoy when it is closed, causing Envoy to reset the upstream connection to the model server.
-type ImmediateResponseAborter struct {
+type ImmediateResponseEvictor struct {
 	// closeOnce tracks which channels have been closed to prevent double-close panics.
 	closeOnce sync.Map // requestID → *sync.Once
 }
 
-// NewImmediateResponseAborter creates an ImmediateResponseAborter.
-func NewImmediateResponseAborter() *ImmediateResponseAborter {
-	return &ImmediateResponseAborter{}
+// NewImmediateResponseEvictor creates an ImmediateResponseEvictor.
+func NewImmediateResponseEvictor() *ImmediateResponseEvictor {
+	return &ImmediateResponseEvictor{}
 }
 
-func (a *ImmediateResponseAborter) Abort(ctx context.Context, item *flowcontrol.EvictionItem) error {
+func (e *ImmediateResponseEvictor) Evict(ctx context.Context, item *flowcontrol.EvictionItem) error {
 	if item.AbortCh == nil {
 		return fmt.Errorf("eviction item %s has no abort channel", item.RequestID)
 	}
 
-	// Use sync.Once to safely close the channel exactly once.
-	once, _ := a.closeOnce.LoadOrStore(item.RequestID, &sync.Once{})
+	once, _ := e.closeOnce.LoadOrStore(item.RequestID, &sync.Once{})
 	once.(*sync.Once).Do(func() {
 		close(item.AbortCh)
 	})
 
-	log.FromContext(ctx).Info("Abort signal sent",
+	log.FromContext(ctx).Info("Eviction signal sent",
 		"requestID", item.RequestID,
 		"priority", item.Priority,
 		"targetURL", item.TargetURL)
 	return nil
 }
+
+// Cleanup removes the sync.Once entry for a request ID to prevent unbounded map growth.
+// Called when a request completes or is untracked.
+func (e *ImmediateResponseEvictor) Cleanup(requestID string) {
+	e.closeOnce.Delete(requestID)
+}

pkg/epp/flowcontrol/eviction/aborter_test.go

@@ -26,68 +26,82 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
 )
 
-func TestImmediateResponseAborter_ClosesChannel(t *testing.T) {
+func TestImmediateResponseEvictor_ClosesChannel(t *testing.T) {
 	t.Parallel()
-	aborter := NewImmediateResponseAborter()
+	evictor := NewImmediateResponseEvictor()
 
 	abortCh := make(chan struct{})
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
 		AbortCh:   abortCh,
 	}
 
-	err := aborter.Abort(context.Background(), item)
+	err := evictor.Evict(context.Background(), item)
 	require.NoError(t, err)
 
-	// Channel should be closed.
 	select {
 	case <-abortCh:
-		// success — channel is closed
 	default:
-		t.Fatal("abort channel should be closed after Abort()")
+		t.Fatal("abort channel should be closed after Evict()")
 	}
 }
 
-func TestImmediateResponseAborter_DoubleAbortSafe(t *testing.T) {
+func TestImmediateResponseEvictor_DoubleEvictSafe(t *testing.T) {
 	t.Parallel()
-	aborter := NewImmediateResponseAborter()
+	evictor := NewImmediateResponseEvictor()
 
 	abortCh := make(chan struct{})
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
 		AbortCh:   abortCh,
 	}
 
-	// First abort should succeed.
-	err := aborter.Abort(context.Background(), item)
+	err := evictor.Evict(context.Background(), item)
 	require.NoError(t, err)
 
-	// Second abort on same request should not panic.
-	err = aborter.Abort(context.Background(), item)
+	// Second evict on same request should not panic.
+	err = evictor.Evict(context.Background(), item)
 	require.NoError(t, err)
 }
 
-func TestImmediateResponseAborter_NilChannel(t *testing.T) {
+func TestImmediateResponseEvictor_NilChannel(t *testing.T) {
 	t.Parallel()
-	aborter := NewImmediateResponseAborter()
+	evictor := NewImmediateResponseEvictor()
 
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
 		AbortCh:   nil,
 	}
 
-	err := aborter.Abort(context.Background(), item)
-	assert.Error(t, err, "Abort with nil channel should return error")
+	err := evictor.Evict(context.Background(), item)
+	assert.Error(t, err, "Evict with nil channel should return error")
 }
 
-func TestNoOpAborter(t *testing.T) {
+func TestImmediateResponseEvictor_Cleanup(t *testing.T) {
 	t.Parallel()
-	aborter := &NoOpAborter{}
+	evictor := NewImmediateResponseEvictor()
+
+	abortCh := make(chan struct{})
+	item := &flowcontrol.EvictionItem{
+		RequestID: "req-1",
+		AbortCh:   abortCh,
+	}
+
+	_ = evictor.Evict(context.Background(), item)
+	evictor.Cleanup("req-1")
+
+	// Cleanup non-existent should not panic.
+	evictor.Cleanup("non-existent")
+}
+
+func TestNoOpEvictor(t *testing.T) {
+	t.Parallel()
+	evictor := &NoOpEvictor{}
 
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
 	}
 
-	err := aborter.Abort(context.Background(), item)
-	assert.NoError(t, err, "NoOpAborter should always succeed")
+	err := evictor.Evict(context.Background(), item)
+	assert.NoError(t, err, "NoOpEvictor should always succeed")
 }

pkg/epp/flowcontrol/eviction/plugin.go

@@ -32,42 +32,42 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/scheduling"
 )
 
-var _ requestcontrol.PreRequest = &Plugin{}
-var _ requestcontrol.ResponseBody = &Plugin{}
+var _ requestcontrol.PreRequest = &RequestEvictor{}
+var _ requestcontrol.ResponseBody = &RequestEvictor{}
 
-// Plugin tracks in-flight requests via RequestControl hooks and provides eviction capability.
-type Plugin struct {
+// RequestEvictor tracks in-flight requests via RequestControl hooks and provides eviction capability.
+type RequestEvictor struct {
 	queue         *EvictionQueue
-	aborter       Aborter
+	evictor       Evictor
 	abortRegistry *AbortRegistry
 }
 
-// NewPlugin creates an eviction plugin with the given policies and aborter.
-func NewPlugin(
+// NewRequestEvictor creates a RequestEvictor with the given policies and evictor.
+func NewRequestEvictor(
 	ordering flowcontrol.EvictionOrderingPolicy,
 	filter flowcontrol.EvictionFilterPolicy,
-	aborter Aborter,
-) *Plugin {
-	return &Plugin{
+	evictor Evictor,
+) *RequestEvictor {
+	return &RequestEvictor{
 		queue:         NewEvictionQueue(ordering, filter),
-		aborter:       aborter,
+		evictor:       evictor,
 		abortRegistry: NewAbortRegistry(),
 	}
 }
 
 // AbortRegistry returns the shared abort registry.
 // The ext_proc Process() goroutine uses this to look up abort channels for dispatched requests.
-func (p *Plugin) AbortRegistry() *AbortRegistry {
+func (p *RequestEvictor) AbortRegistry() *AbortRegistry {
 	return p.abortRegistry
 }
 
-func (p *Plugin) TypedName() plugin.TypedName {
+func (p *RequestEvictor) TypedName() plugin.TypedName {
 	return plugin.TypedName{Type: "EvictionPlugin", Name: "eviction"}
 }
 
 // PreRequest is called after scheduling, before the request reaches the model server.
 // It tracks the request and, if the filter policy accepts it, adds it to the eviction queue.
-func (p *Plugin) PreRequest(
+func (p *RequestEvictor) PreRequest(
 	ctx context.Context,
 	request *scheduling.LLMRequest,
 	result *scheduling.SchedulingResult,
@@ -121,7 +121,7 @@ func (p *Plugin) PreRequest(
 
 // ResponseBody is called for every response data chunk (streaming) or once (non-streaming).
 // On the final call (EndOfStream == true), it removes the request from tracking and the eviction queue.
-func (p *Plugin) ResponseBody(
+func (p *RequestEvictor) ResponseBody(
 	ctx context.Context,
 	request *scheduling.LLMRequest,
 	response *requestcontrol.Response,
@@ -151,7 +151,7 @@ func (p *Plugin) ResponseBody(
 // Each request is only removed from tracking after a successful abort. If the abort fails,
 // the request remains in the queue for a future eviction attempt.
 // Returns the request IDs that were successfully aborted.
-func (p *Plugin) EvictN(ctx context.Context, n int) ([]string, error) {
+func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) {
 	logger := log.FromContext(ctx)
 	aborted := make([]string, 0, n)
 
@@ -162,8 +162,8 @@ func (p *Plugin) EvictN(ctx context.Context, n int) ([]string, error) {
 		}
 		item := items[0]
 
-		if err := p.aborter.Abort(ctx, item); err != nil {
-			logger.Error(err, "Failed to abort request, re-tracking", "requestID", item.RequestID, "targetURL", item.TargetURL)
+		if err := p.evictor.Evict(ctx, item); err != nil {
+			logger.Error(err, "Failed to evict request, re-tracking", "requestID", item.RequestID, "targetURL", item.TargetURL)
 			p.queue.Track(item)
 			continue
 		}
@@ -177,6 +177,6 @@ func (p *Plugin) EvictN(ctx context.Context, n int) ([]string, error) {
 }
 
 // Stats returns the current in-flight and evictable request counts.
-func (p *Plugin) Stats() (inFlight int, evictable int) {
+func (p *RequestEvictor) Stats() (inFlight int, evictable int) {
 	return p.queue.InFlightLen(), p.queue.EvictableLen()
 }