fix(cag): extract applyWarmup/acquireScheduledBackend helpers, add race-free tests

- proxy.go: extract applyWarmup + acquireScheduledBackend to reduce ServeHTTP complexity - proxy_test.go: TestProxyQueueFull, TestProxyCacheHitLogging (AC5/AC6/AC7 HTTP-level coverage) - scheduler_test.go: TestBackendAcquireCancelledContext + runtime.Gosched spin-wait (no time.Sleep) - router_test.go: TestAuthenticate_MethodCapabilityOverrideDirect with real RS256 JWT + JWKS server
2026-09-27 18:57:04 +08:00
parent a731185588
commit d703ace2c6
4 changed files with 343 additions and 93 deletions
@@ -2,6 +2,7 @@
 package proxy

 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -468,56 +469,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var selectedBackend *Backend
 	cacheKey := r.Header.Get("X-Poimen-Cache-Key")
 	if sched := h.schedulers[route.ModelName]; sched != nil {
-		selectedBackend = sched.Select(cacheKey)
-
-		// Warmup: sleep BEFORE queue entry so agents arriving in that window win slots first.
-		if r.Header.Get("X-Poimen-Warmup") == "true" {
-			r.Header.Del("X-Poimen-Warmup")
-			r.Header.Set("X-Priority", "low")
-			time.Sleep(h.warmupDelay)
-			observability.Global().IncrWarmupDelayed()
+		var ok bool
+		selectedBackend, route, ok = h.acquireScheduledBackend(r.Context(), w, r, sched, route, cacheKey)
+		if !ok {
+			return
 		}
-
 		if selectedBackend != nil {
-			if err := selectedBackend.Acquire(r.Context()); err != nil {
-				if errors.Is(err, ErrQueueFull) {
-					observability.Global().IncrQueueFull()
-					logging.Warnf("backend queue full", map[string]string{
-						"backend": selectedBackend.Address,
-						"path":    r.URL.Path,
-						"model":   route.ModelName,
-					})
-					w.WriteHeader(http.StatusServiceUnavailable)
-					fmt.Fprintf(w, "queue full")
-				} else {
-					logging.Warnf("backend acquire failed", map[string]string{
-						"backend": selectedBackend.Address,
-						"path":    r.URL.Path,
-						"model":   route.ModelName,
-						"err":     err.Error(),
-					})
-					w.WriteHeader(http.StatusServiceUnavailable)
-					fmt.Fprintf(w, "upstream busy")
-				}
-				return
-			}
 			defer selectedBackend.Release()
-
-			// Patch route to selected backend if it differs from the config default.
-			if selectedBackend.Address != route.Upstream.Address {
-				patchedCfg := *route.Upstream
-				patchedCfg.Address = selectedBackend.Address
-				patchedURL, _ := url.Parse("http://" + selectedBackend.Address)
-				route = &Route{
-					Name:      route.Name,
-					ModelName: route.ModelName,
-					Upstream:  &patchedCfg,
-					Transport: h.getOrCreateTransport(selectedBackend.Address, &patchedCfg),
-					Director: func(req *http.Request) {
-						directorFunc(req, patchedURL, &patchedCfg)
-					},
-				}
-			}
 		}
 	}

@@ -586,6 +544,68 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {



+// applyWarmup strips the X-Poimen-Warmup header, delays the request, and records
+// the metric. It is a no-op when the header is absent or not "true".
+func (h *Handler) applyWarmup(r *http.Request) {
+	if r.Header.Get("X-Poimen-Warmup") != "true" {
+		return
+	}
+	r.Header.Del("X-Poimen-Warmup")
+	r.Header.Set("X-Priority", "low")
+	time.Sleep(h.warmupDelay)
+	observability.Global().IncrWarmupDelayed()
+}
+
+// acquireScheduledBackend selects a backend from sched, applies warmup
+// deprioritisation, acquires a concurrency slot, and patches route when the
+// selected backend differs from the config default. It returns the backend
+// to Release (caller must defer), the (possibly patched) route, and ok=true.
+// If the slot cannot be acquired it writes a 503 and returns (nil, nil, false).
+func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool) {
+	backend := sched.Select(cacheKey)
+	h.applyWarmup(r)
+	if backend == nil {
+		return nil, route, true
+	}
+	if err := backend.Acquire(ctx); err != nil {
+		if errors.Is(err, ErrQueueFull) {
+			observability.Global().IncrQueueFull()
+			logging.Warnf("backend queue full", map[string]string{
+				"backend": backend.Address,
+				"path":    r.URL.Path,
+				"model":   route.ModelName,
+			})
+			w.WriteHeader(http.StatusServiceUnavailable)
+			fmt.Fprintf(w, "queue full")
+		} else {
+			logging.Warnf("backend acquire failed", map[string]string{
+				"backend": backend.Address,
+				"path":    r.URL.Path,
+				"model":   route.ModelName,
+				"err":     err.Error(),
+			})
+			w.WriteHeader(http.StatusServiceUnavailable)
+			fmt.Fprintf(w, "upstream busy")
+		}
+		return nil, nil, false
+	}
+	if backend.Address != route.Upstream.Address {
+		patchedCfg := *route.Upstream
+		patchedCfg.Address = backend.Address
+		patchedURL, _ := url.Parse("http://" + backend.Address)
+		route = &Route{
+			Name:      route.Name,
+			ModelName: route.ModelName,
+			Upstream:  &patchedCfg,
+			Transport: h.getOrCreateTransport(backend.Address, &patchedCfg),
+			Director: func(req *http.Request) {
+				directorFunc(req, patchedURL, &patchedCfg)
+			},
+		}
+	}
+	return backend, route, true
+}
+
 // getValidModels returns a sorted list of all configured model names.
 func (h *Handler) getValidModels() []string {
 	var models []string
@@ -5,11 +5,13 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"strings"
 	"testing"
 	"time"

 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
 )

 func TestProxyBasic(t *testing.T) {
@@ -415,6 +417,140 @@ func TestProxyPreservesBody(t *testing.T) {
 	}
 }

+// TestProxyQueueFull verifies that a 503 is returned and the queue_full metric
+// is incremented when the backend queue is at capacity.
+//
+// Setup: maxSlots=1, maxQueue=1. Request 1 holds the slot. Request 2 fills the
+// queue (blocks waiting for slot). Request 3 gets ErrQueueFull → 503.
+func TestProxyQueueFull(t *testing.T) {
+	t.Setenv("VLLM_MAX_NUM_SEQS", "1")    // 1 concurrent slot
+	t.Setenv("VLLM_MAX_QUEUE_DEPTH", "1") // queue capacity 1
+	t.Setenv("WARMUP_DELAY_MS", "0")
+	observability.Global().Reset()
+
+	// Upstream blocks until released; buffers one signal when it first starts.
+	slotHeld := make(chan struct{}, 1)
+	release := make(chan struct{})
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case slotHeld <- struct{}{}:
+		default:
+		}
+		<-release
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer upstream.Close()
+
+	upstreamAddr := strings.TrimPrefix(upstream.URL, "http://")
+	cfg := &config.Config{
+		Routes: make(map[string]*config.Route),
+		Models: map[string]*config.ModelUpstream{
+			"q-model": {
+				Name:    "q-model",
+				Address: upstreamAddr,
+			},
+		},
+	}
+	h := New(cfg)
+	defer h.Close()
+
+	srv := httptest.NewServer(h)
+	defer srv.Close()
+
+	mkReq := func() *http.Request {
+		body := strings.NewReader(`{"model":"q-model","messages":[]}`)
+		req, _ := http.NewRequest("POST", srv.URL+"/v1/chat/completions", body)
+		req.Header.Set("Content-Type", "application/json")
+		return req
+	}
+
+	// Request 1: acquires the only slot, reaches upstream, blocks there.
+	r1ch := make(chan *http.Response, 1)
+	go func() {
+		resp, _ := http.DefaultClient.Do(mkReq())
+		r1ch <- resp
+	}()
+	<-slotHeld // slot is acquired; upstream is holding request 1
+
+	// Request 2: enters the queue (capacity 1), blocks waiting for slot.
+	r2ch := make(chan *http.Response, 1)
+	go func() {
+		resp, _ := http.DefaultClient.Do(mkReq())
+		r2ch <- resp
+	}()
+	// Spin until request 2 enters the backend queue channel.
+	for len(h.schedulers["q-model"].backends[0].queue) == 0 {
+		runtime.Gosched()
+	}
+
+	// Request 3: queue is full → expect 503 and metric increment.
+	resp3, err := http.DefaultClient.Do(mkReq())
+	if err != nil {
+		t.Fatalf("request 3 failed: %v", err)
+	}
+	defer resp3.Body.Close()
+
+	if resp3.StatusCode != http.StatusServiceUnavailable {
+		t.Errorf("expected 503 queue full, got %d", resp3.StatusCode)
+	}
+	if got := observability.Global().GetQueueFull(); got != 1 {
+		t.Errorf("expected GetQueueFull()=1, got %d", got)
+	}
+
+	// Cleanup: unblock upstream so goroutines can finish.
+	close(release)
+	if r := <-r1ch; r != nil {
+		r.Body.Close()
+	}
+	if r := <-r2ch; r != nil {
+		r.Body.Close()
+	}
+}
+
+// TestProxyCacheHitLogging verifies that when a response contains the
+// vllm-num-cached-tokens header and a cache key was set, the prefix cache
+// hit token count is incremented in observability metrics.
+func TestProxyCacheHitLogging(t *testing.T) {
+	observability.Global().Reset()
+
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("vllm-num-cached-tokens", "128")
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer upstream.Close()
+
+	upstreamAddr := strings.TrimPrefix(upstream.URL, "http://")
+	cfg := &config.Config{
+		Routes: make(map[string]*config.Route),
+		Models: map[string]*config.ModelUpstream{
+			"cache-model": {
+				Name:    "cache-model",
+				Address: upstreamAddr,
+			},
+		},
+	}
+	h := New(cfg)
+	defer h.Close()
+
+	srv := httptest.NewServer(h)
+	defer srv.Close()
+
+	body := strings.NewReader(`{"model":"cache-model","messages":[]}`)
+	req, _ := http.NewRequest("POST", srv.URL+"/v1/chat/completions", body)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Poimen-Cache-Key", "testkey123")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	resp.Body.Close()
+
+	if got := observability.Global().GetPrefixCacheHitTokens(); got != 128 {
+		t.Errorf("expected PrefixCacheHitTokens=128, got %d", got)
+	}
+}
+
 // TestWarmupHeaderHandling verifies that X-Poimen-Warmup: true is stripped from
 // the upstream request and replaced with X-Priority: low, while a request without
 // X-Poimen-Warmup reaches upstream without either header.
@@ -3,6 +3,7 @@ package proxy
 import (
 	"context"
 	"errors"
+	"runtime"
 	"sync"
 	"testing"
 	"time"
@@ -23,8 +24,10 @@ func TestBackendAcquireBlocksThenUnblocksOnRelease(t *testing.T) {
 		g2Done <- b.Acquire(context.Background())
 	}()

-	// Give G2 time to block.
-	time.Sleep(20 * time.Millisecond)
+	// Wait until G2 enters the queue channel (proves Acquire is blocking on slots).
+	for len(b.queue) == 0 {
+		runtime.Gosched()
+	}

 	select {
 	case err := <-g2Done:
@@ -119,8 +122,10 @@ func TestBackendAcquireCancelledContext(t *testing.T) {
 		g2Done <- b.Acquire(ctx)
 	}()

-	// Give G2 time to enter the queue and block waiting for a slot.
-	time.Sleep(20 * time.Millisecond)
+	// Wait until G2 enters the queue channel.
+	for len(b.queue) == 0 {
+		runtime.Gosched()
+	}

 	// Verify G2 is queued (queue channel has one item).
 	if qd := len(b.queue); qd != 1 {