Refactor Process() loop: single reader goroutine, state machine eviction, evictor cleanup

- Replace per-message recvOrAbort goroutine with single reader goroutine
  for the stream lifetime, using nil channel select pattern
- Remove errEvicted sentinel; eviction is a state transition (RequestEvicted)
  handled by updateStateAndSendIfNeeded, not an error
- Move ImmediateResponse send from inline code into the state machine
- Add EvictorWithCleanup interface and cleanupRequest helper to prevent
  ImmediateResponseEvictor.closeOnce map from growing unbounded

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

Author: RishabhSaini

…p/flowcontrol/eviction/abort_registry.go …lowcontrol/eviction/eviction_registry.gopkg/epp/flowcontrol/eviction/abort_registry.go renamed to pkg/epp/flowcontrol/eviction/eviction_registry.go pkg/epp/flowcontrol/eviction/abort_registry.go renamed to pkg/epp/flowcontrol/eviction/eviction_registry.go

@@ -18,45 +18,45 @@ package eviction
 
 import "sync"
 
-// AbortRegistry is a shared registry that maps request IDs to abort channels.
+// EvictionRegistry is a shared registry that maps request IDs to eviction channels.
 // It bridges the eviction plugin (which decides what to evict) and the ext_proc Process()
 // goroutine (which owns the stream needed to send ImmediateResponse).
 //
 // Lifecycle:
-//   - PreRequest: plugin creates an abort channel and registers it via Register().
+//   - PreRequest: plugin creates an eviction channel and registers it via Register().
 //   - Process(): after HandleRequest returns, looks up the channel via Get() and selects on it.
-//   - EvictN: aborter closes the channel via the EvictionItem.AbortCh reference.
+//   - EvictN: evictor closes the channel via the EvictionItem.EvictCh reference.
 //   - Process() defer: removes the channel via Deregister().
 //
 // All methods are goroutine-safe.
-type AbortRegistry struct {
+type EvictionRegistry struct {
 	mu       sync.RWMutex
-	channels map[string]chan struct{} // requestID → abort channel
+	channels map[string]chan struct{} // requestID → eviction channel
 }
 
-// NewAbortRegistry creates a new AbortRegistry.
-func NewAbortRegistry() *AbortRegistry {
-	return &AbortRegistry{
+// NewEvictionRegistry creates a new EvictionRegistry.
+func NewEvictionRegistry() *EvictionRegistry {
+	return &EvictionRegistry{
 		channels: make(map[string]chan struct{}),
 	}
 }
 
-// Register stores an abort channel for the given request ID.
-func (r *AbortRegistry) Register(requestID string, ch chan struct{}) {
+// Register stores an eviction channel for the given request ID.
+func (r *EvictionRegistry) Register(requestID string, ch chan struct{}) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.channels[requestID] = ch
 }
 
-// Get returns the abort channel for the given request ID, or nil if not found.
-func (r *AbortRegistry) Get(requestID string) chan struct{} {
+// Get returns the eviction channel for the given request ID, or nil if not found.
+func (r *EvictionRegistry) Get(requestID string) chan struct{} {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 	return r.channels[requestID]
 }
 
-// Deregister removes the abort channel for the given request ID.
-func (r *AbortRegistry) Deregister(requestID string) {
+// Deregister removes the eviction channel for the given request ID.
+func (r *EvictionRegistry) Deregister(requestID string) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	delete(r.channels, requestID)

…wcontrol/eviction/abort_registry_test.go …ntrol/eviction/eviction_registry_test.gopkg/epp/flowcontrol/eviction/abort_registry_test.go renamed to pkg/epp/flowcontrol/eviction/eviction_registry_test.go pkg/epp/flowcontrol/eviction/abort_registry_test.go renamed to pkg/epp/flowcontrol/eviction/eviction_registry_test.go

@@ -24,9 +24,9 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestAbortRegistry_RegisterAndGet(t *testing.T) {
+func TestEvictionRegistry_RegisterAndGet(t *testing.T) {
 	t.Parallel()
-	r := NewAbortRegistry()
+	r := NewEvictionRegistry()
 
 	ch := make(chan struct{})
 	r.Register("req-1", ch)
@@ -37,9 +37,9 @@ func TestAbortRegistry_RegisterAndGet(t *testing.T) {
 	assert.Nil(t, r.Get("non-existent"), "Get for unregistered ID should return nil")
 }
 
-func TestAbortRegistry_Deregister(t *testing.T) {
+func TestEvictionRegistry_Deregister(t *testing.T) {
 	t.Parallel()
-	r := NewAbortRegistry()
+	r := NewEvictionRegistry()
 
 	ch := make(chan struct{})
 	r.Register("req-1", ch)
@@ -51,9 +51,9 @@ func TestAbortRegistry_Deregister(t *testing.T) {
 	r.Deregister("non-existent")
 }
 
-func TestAbortRegistry_Concurrency(t *testing.T) {
+func TestEvictionRegistry_Concurrency(t *testing.T) {
 	t.Parallel()
-	r := NewAbortRegistry()
+	r := NewEvictionRegistry()
 
 	const goroutines = 10
 	const opsPerGoroutine = 100

pkg/epp/flowcontrol/eviction/aborter.go pkg/epp/flowcontrol/eviction/evictor.gopkg/epp/flowcontrol/eviction/aborter.go renamed to pkg/epp/flowcontrol/eviction/evictor.go

@@ -43,7 +43,7 @@ func (e *NoOpEvictor) Evict(ctx context.Context, item *flowcontrol.EvictionItem)
 	return nil
 }
 
-// ImmediateResponseEvictor evicts requests by closing the EvictionItem's AbortCh.
+// ImmediateResponseEvictor evicts requests by closing the EvictionItem's EvictCh.
 // 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 ImmediateResponseEvictor struct {
@@ -57,13 +57,13 @@ func NewImmediateResponseEvictor() *ImmediateResponseEvictor {
 }
 
 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)
+	if item.EvictCh == nil {
+		return fmt.Errorf("eviction item %s has no eviction channel", item.RequestID)
 	}
 
 	once, _ := e.closeOnce.LoadOrStore(item.RequestID, &sync.Once{})
 	once.(*sync.Once).Do(func() {
-		close(item.AbortCh)
+		close(item.EvictCh)
 	})
 
 	log.FromContext(ctx).Info("Eviction signal sent",
@@ -72,3 +72,9 @@ func (e *ImmediateResponseEvictor) Evict(ctx context.Context, item *flowcontrol.
 		"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)
+}

…epp/flowcontrol/eviction/aborter_test.go …epp/flowcontrol/eviction/evictor_test.gopkg/epp/flowcontrol/eviction/aborter_test.go renamed to pkg/epp/flowcontrol/eviction/evictor_test.go pkg/epp/flowcontrol/eviction/aborter_test.go renamed to pkg/epp/flowcontrol/eviction/evictor_test.go

@@ -30,30 +30,30 @@ func TestImmediateResponseEvictor_ClosesChannel(t *testing.T) {
 	t.Parallel()
 	evictor := NewImmediateResponseEvictor()
 
-	abortCh := make(chan struct{})
+	evictCh := make(chan struct{})
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
-		AbortCh:   abortCh,
+		EvictCh:   evictCh,
 	}
 
 	err := evictor.Evict(context.Background(), item)
 	require.NoError(t, err)
 
 	select {
-	case <-abortCh:
+	case <-evictCh:
 	default:
-		t.Fatal("abort channel should be closed after Evict()")
+		t.Fatal("eviction channel should be closed after Evict()")
 	}
 }
 
 func TestImmediateResponseEvictor_DoubleEvictSafe(t *testing.T) {
 	t.Parallel()
 	evictor := NewImmediateResponseEvictor()
 
-	abortCh := make(chan struct{})
+	evictCh := make(chan struct{})
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
-		AbortCh:   abortCh,
+		EvictCh:   evictCh,
 	}
 
 	err := evictor.Evict(context.Background(), item)
@@ -70,13 +70,48 @@ func TestImmediateResponseEvictor_NilChannel(t *testing.T) {
 
 	item := &flowcontrol.EvictionItem{
 		RequestID: "req-1",
-		AbortCh:   nil,
+		EvictCh:   nil,
 	}
 
 	err := evictor.Evict(context.Background(), item)
 	assert.Error(t, err, "Evict with nil channel should return error")
 }
 
+func TestImmediateResponseEvictor_Cleanup(t *testing.T) {
+	t.Parallel()
+	evictor := NewImmediateResponseEvictor()
+
+	evictCh := make(chan struct{})
+	item := &flowcontrol.EvictionItem{
+		RequestID: "req-1",
+		EvictCh:   evictCh,
+	}
+
+	_ = evictor.Evict(context.Background(), item)
+
+	// Cleanup should remove the sync.Once entry.
+	evictor.Cleanup("req-1")
+
+	// After cleanup, a new Evict on the same requestID with a new channel should work
+	// (the old sync.Once is gone, so a new one will be created).
+	evictCh2 := make(chan struct{})
+	item2 := &flowcontrol.EvictionItem{
+		RequestID: "req-1",
+		EvictCh:   evictCh2,
+	}
+	err := evictor.Evict(context.Background(), item2)
+	require.NoError(t, err)
+
+	select {
+	case <-evictCh2:
+	default:
+		t.Fatal("new channel should be closed after Evict post-Cleanup")
+	}
+
+	// Cleanup non-existent should not panic.
+	evictor.Cleanup("non-existent")
+}
+
 func TestNoOpEvictor(t *testing.T) {
 	t.Parallel()
 	evictor := &NoOpEvictor{}

pkg/epp/flowcontrol/eviction/plugin.go

@@ -37,9 +37,9 @@ var _ requestcontrol.ResponseBody = &RequestEvictor{}
 
 // RequestEvictor tracks in-flight requests via RequestControl hooks and provides eviction capability.
 type RequestEvictor struct {
-	queue         *EvictionQueue
-	evictor       Evictor
-	abortRegistry *AbortRegistry
+	queue            *EvictionQueue
+	evictor          Evictor
+	evictionRegistry *EvictionRegistry
 }
 
 // NewRequestEvictor creates a RequestEvictor with the given policies and evictor.
@@ -49,16 +49,16 @@ func NewRequestEvictor(
 	evictor Evictor,
 ) *RequestEvictor {
 	return &RequestEvictor{
-		queue:         NewEvictionQueue(ordering, filter),
-		evictor:       evictor,
-		abortRegistry: NewAbortRegistry(),
+		queue:            NewEvictionQueue(ordering, filter),
+		evictor:          evictor,
+		evictionRegistry: NewEvictionRegistry(),
 	}
 }
 
-// AbortRegistry returns the shared abort registry.
-// The ext_proc Process() goroutine uses this to look up abort channels for dispatched requests.
-func (p *RequestEvictor) AbortRegistry() *AbortRegistry {
-	return p.abortRegistry
+// EvictionRegistry returns the shared eviction registry.
+// The ext_proc Process() goroutine uses this to look up eviction channels for dispatched requests.
+func (p *RequestEvictor) EvictionRegistry() *EvictionRegistry {
+	return p.evictionRegistry
 }
 
 func (p *RequestEvictor) TypedName() plugin.TypedName {
@@ -88,7 +88,7 @@ func (p *RequestEvictor) PreRequest(
 		return
 	}
 
-	abortCh := make(chan struct{})
+	evictCh := make(chan struct{})
 
 	item := &flowcontrol.EvictionItem{
 		RequestID:      requestID,
@@ -97,19 +97,18 @@ func (p *RequestEvictor) PreRequest(
 		TargetURL:      "http://" + net.JoinHostPort(metadata.GetIPAddress(), metadata.GetPort()),
 		Request:        request,
 		TargetEndpoint: metadata,
-		AbortCh:        abortCh,
+		EvictCh:        evictCh,
 	}
 
 	p.queue.Track(item)
-	p.abortRegistry.Register(requestID, abortCh)
+	p.evictionRegistry.Register(requestID, evictCh)
 
 	// Bind untrack to the request context's lifetime as a safety net.
 	// If the client disconnects and ResponseBody(EndOfStream) never fires,
 	// ctx.Done() ensures the request is still cleaned up. Untrack is idempotent.
 	go func() {
 		<-ctx.Done()
-		p.queue.Untrack(requestID)
-		p.abortRegistry.Deregister(requestID)
+		p.cleanupRequest(requestID)
 	}()
 
 	log.FromContext(ctx).V(logutil.DEBUG).Info("Tracked in-flight request",
@@ -138,8 +137,7 @@ func (p *RequestEvictor) ResponseBody(
 		return
 	}
 
-	p.queue.Untrack(requestID)
-	p.abortRegistry.Deregister(requestID)
+	p.cleanupRequest(requestID)
 
 	log.FromContext(ctx).V(logutil.DEBUG).Info("Untracked completed request",
 		"requestID", requestID,
@@ -148,12 +146,12 @@ func (p *RequestEvictor) ResponseBody(
 }
 
 // EvictN attempts to evict up to n requests from the eviction queue.
-// Each request is only removed from tracking after a successful abort. If the abort fails,
+// Each request is only removed from tracking after a successful eviction. If the eviction fails,
 // the request remains in the queue for a future eviction attempt.
-// Returns the request IDs that were successfully aborted.
+// Returns the request IDs that were successfully evicted.
 func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) {
 	logger := log.FromContext(ctx)
-	aborted := make([]string, 0, n)
+	evicted := make([]string, 0, n)
 
 	for range n {
 		items := p.queue.PopN(1)
@@ -167,16 +165,33 @@ func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) {
 			p.queue.Track(item)
 			continue
 		}
-		aborted = append(aborted, item.RequestID)
+		evicted = append(evicted, item.RequestID)
 	}
 
-	if len(aborted) > 0 {
-		logger.Info("Eviction complete", "requested", n, "aborted", len(aborted))
+	if len(evicted) > 0 {
+		logger.Info("Eviction complete", "requested", n, "evicted", len(evicted))
 	}
-	return aborted, nil
+	return evicted, nil
 }
 
 // Stats returns the current in-flight and evictable request counts.
 func (p *RequestEvictor) Stats() (inFlight int, evictable int) {
 	return p.queue.InFlightLen(), p.queue.EvictableLen()
 }
+
+// cleanupRequest removes a request from all tracking structures.
+// If the evictor supports cleanup (e.g., ImmediateResponseEvictor), it also
+// cleans up evictor-internal state to prevent unbounded map growth.
+func (p *RequestEvictor) cleanupRequest(requestID string) {
+	p.queue.Untrack(requestID)
+	p.evictionRegistry.Deregister(requestID)
+	if c, ok := p.evictor.(EvictorWithCleanup); ok {
+		c.Cleanup(requestID)
+	}
+}
+
+// EvictorWithCleanup is an optional interface for evictors that maintain per-request state
+// that needs to be cleaned up when a request completes or is untracked.
+type EvictorWithCleanup interface {
+	Cleanup(requestID string)
+}