Add the discovery HTTP API: list, lease, release, report over the manager cache

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-09 15:26:11 +02:00
parent f6d50e4744
commit 4aa3d47e3c
6 changed files with 950 additions and 8 deletions
--- a/internal/discovery/handlers.go
+++ b/internal/discovery/handlers.go
@@ -0,0 +1,235 @@
+package discovery
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"slices"
+	"strings"
+	"time"
+
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease"
+)
+
+// proxyView is the wire shape of one proxy in list and lease responses.
+type proxyView struct {
+	ID            string            `json:"id"` // namespace/name
+	IP            string            `json:"ip"`
+	Port          int32             `json:"port"`
+	Attributes    map[string]string `json:"attributes,omitempty"`
+	Phase         string            `json:"phase"`
+	Healthy       bool              `json:"healthy"`
+	LatencyMillis int64             `json:"latencyMillis"`
+	ActiveLeases  int               `json:"activeLeases"`
+	MaxLeases     int32             `json:"maxLeases"`
+}
+
+func viewOf(p *crawlv1alpha1.Proxy, activeLeases int) proxyView {
+	return proxyView{
+		ID:            client.ObjectKeyFromObject(p).String(),
+		IP:            p.EffectiveHost(),
+		Port:          p.EffectivePort(),
+		Attributes:    p.Spec.Attributes,
+		Phase:         string(p.Status.Phase),
+		Healthy:       isHealthy(p),
+		LatencyMillis: p.Status.LatencyMillis,
+		ActiveLeases:  activeLeases,
+		MaxLeases:     p.MaxLeasesOrDefault(),
+	}
+}
+
+func isHealthy(p *crawlv1alpha1.Proxy) bool {
+	return apimeta.IsStatusConditionTrue(p.Status.Conditions, crawlv1alpha1.ConditionHealthy)
+}
+
+// matchesAttributes is spec.attributes equality: every selector pair must
+// be present verbatim.
+func matchesAttributes(p *crawlv1alpha1.Proxy, selector map[string]string) bool {
+	for k, v := range selector {
+		if p.Spec.Attributes[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+// GET /v1/proxies?attr.<key>=<value>&healthy=true|false
+func (s *Server) handleListProxies(w http.ResponseWriter, r *http.Request) {
+	selector := map[string]string{}
+	var healthyFilter *bool
+	for key, values := range r.URL.Query() {
+		switch {
+		case key == "healthy":
+			switch values[0] {
+			case "true":
+				healthyFilter = ptr(true)
+			case "false":
+				healthyFilter = ptr(false)
+			default:
+				writeError(w, http.StatusBadRequest, "invalid_query",
+					fmt.Sprintf("healthy must be true or false, got %q", values[0]))
+				return
+			}
+		case strings.HasPrefix(key, "attr."):
+			selector[strings.TrimPrefix(key, "attr.")] = values[0]
+		}
+	}
+
+	var list crawlv1alpha1.ProxyList
+	if err := s.Reader.List(r.Context(), &list); err != nil {
+		s.log.Error(err, "listing proxies")
+		writeError(w, http.StatusInternalServerError, "internal", "listing proxies failed")
+		return
+	}
+	counts := s.Store.Counts()
+
+	views := []proxyView{}
+	for i := range list.Items {
+		p := &list.Items[i]
+		if !p.DeletionTimestamp.IsZero() || !matchesAttributes(p, selector) {
+			continue
+		}
+		v := viewOf(p, counts[client.ObjectKeyFromObject(p).String()])
+		if healthyFilter != nil && v.Healthy != *healthyFilter {
+			continue
+		}
+		views = append(views, v)
+	}
+	slices.SortFunc(views, func(a, b proxyView) int { return strings.Compare(a.ID, b.ID) })
+
+	writeJSON(w, http.StatusOK, map[string]any{"proxies": views, "count": len(views)})
+}
+
+type leaseRequest struct {
+	Selector   map[string]string `json:"selector"`
+	TTLSeconds int64             `json:"ttlSeconds"`
+	Target     string            `json:"target"`
+}
+
+type leaseResponse struct {
+	LeaseID    string    `json:"leaseID"`
+	Proxy      proxyView `json:"proxy"`
+	ExpiresAt  time.Time `json:"expiresAt"`
+	TTLSeconds int64     `json:"ttlSeconds"`
+}
+
+// POST /v1/leases
+func (s *Server) handleAcquireLease(w http.ResponseWriter, r *http.Request) {
+	var req leaseRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_body", err.Error())
+		return
+	}
+
+	ttl := time.Duration(req.TTLSeconds) * time.Second
+	if req.TTLSeconds == 0 {
+		ttl = defaultTTL
+	}
+	if ttl < 0 || ttl > s.MaxLeaseTTL {
+		writeError(w, http.StatusBadRequest, "invalid_ttl",
+			fmt.Sprintf("ttlSeconds must be between 1 and %d", int64(s.MaxLeaseTTL.Seconds())))
+		return
+	}
+
+	var list crawlv1alpha1.ProxyList
+	if err := s.Reader.List(r.Context(), &list); err != nil {
+		s.log.Error(err, "listing proxies for lease")
+		writeError(w, http.StatusInternalServerError, "internal", "listing proxies failed")
+		return
+	}
+
+	// The store gets only healthy, live candidates; unhealthy matches are
+	// counted here because the store never sees them.
+	var candidates []lease.Candidate
+	byKey := map[string]*crawlv1alpha1.Proxy{}
+	unhealthy := 0
+	for i := range list.Items {
+		p := &list.Items[i]
+		if !p.DeletionTimestamp.IsZero() || !matchesAttributes(p, req.Selector) {
+			continue
+		}
+		if !isHealthy(p) {
+			unhealthy++
+			continue
+		}
+		key := client.ObjectKeyFromObject(p).String()
+		byKey[key] = p
+		candidates = append(candidates, lease.Candidate{
+			Proxy:     key,
+			MaxLeases: p.MaxLeasesOrDefault(),
+			Latency:   time.Duration(p.Status.LatencyMillis) * time.Millisecond,
+		})
+	}
+
+	granted, stats, err := s.Store.Acquire(r.Context(), lease.AcquireRequest{
+		Candidates: candidates,
+		Target:     req.Target,
+		TTL:        ttl,
+	})
+	if errors.Is(err, lease.ErrNoMatch) {
+		writeJSON(w, http.StatusConflict, map[string]any{
+			"error":      "no_match",
+			"message":    "no healthy proxy with free capacity matched the selector",
+			"considered": stats.Considered + unhealthy,
+			"atCapacity": stats.AtCapacity,
+			"inCooldown": stats.InCooldown,
+			"unhealthy":  unhealthy,
+		})
+		return
+	}
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_request", err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusCreated, leaseResponse{
+		LeaseID:    granted.ID,
+		Proxy:      viewOf(byKey[granted.Proxy], s.Store.Counts()[granted.Proxy]),
+		ExpiresAt:  granted.ExpiresAt,
+		TTLSeconds: int64(ttl.Seconds()),
+	})
+}
+
+// DELETE /v1/leases/{id} — early release, always 204: releasing an unknown
+// or already expired lease is not an error.
+func (s *Server) handleReleaseLease(w http.ResponseWriter, r *http.Request) {
+	s.Store.Release(r.Context(), r.PathValue("id"))
+	w.WriteHeader(http.StatusNoContent)
+}
+
+type reportRequest struct {
+	Result string `json:"result"`
+	Target string `json:"target"`
+}
+
+// POST /v1/leases/{id}/report
+func (s *Server) handleReportLease(w http.ResponseWriter, r *http.Request) {
+	var req reportRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_body", err.Error())
+		return
+	}
+	result, ok := lease.ParseResult(req.Result)
+	if !ok {
+		writeError(w, http.StatusBadRequest, "invalid_result",
+			fmt.Sprintf("result must be one of ok, rate_limited, banned; got %q", req.Result))
+		return
+	}
+	if err := s.Store.Report(r.Context(), r.PathValue("id"), result, req.Target); err != nil {
+		if errors.Is(err, lease.ErrUnknownLease) {
+			writeError(w, http.StatusNotFound, "unknown_lease", "no such lease")
+			return
+		}
+		s.log.Error(err, "reporting lease", "leaseID", r.PathValue("id"))
+		writeError(w, http.StatusInternalServerError, "internal", "report failed")
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func ptr[T any](v T) *T { return &v }
--- a/internal/discovery/server.go
+++ b/internal/discovery/server.go
@@ -0,0 +1,225 @@
+// Package discovery implements the HTTP API crawler clients use to find
+// and lease proxies: list healthy proxies filtered by attributes, acquire a
+// TTL-based lease, release it early, and report how a target treated the
+// proxy. Reads go through the manager's informer cache; lease state lives
+// in the injected store.
+package discovery
+
+import (
+	"context"
+	"crypto/subtle"
+	"encoding/json"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-logr/logr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease"
+)
+
+// LeaseStore is what the handlers need from a lease backend. Defined here,
+// consumer-side, so a CRD- or Redis-backed store can replace the in-memory
+// one (which internal/lease's *Store satisfies) without touching handlers.
+type LeaseStore interface {
+	Acquire(ctx context.Context, req lease.AcquireRequest) (*lease.Lease, lease.AcquireStats, error)
+	Release(ctx context.Context, id string)
+	Report(ctx context.Context, id string, result lease.Result, target string) error
+	Counts() map[string]int
+}
+
+const (
+	defaultAddr     = ":8090"
+	defaultTTL      = 5 * time.Minute
+	defaultMaxTTL   = time.Hour
+	maxBodyBytes    = 64 << 10
+	shutdownGrace   = 10 * time.Second
+	readHeadTimeout = 5 * time.Second
+)
+
+// Server serves the discovery API as a manager Runnable.
+type Server struct {
+	// Reader lists Proxies from the manager's cache.
+	Reader client.Reader
+	// Store is the lease backend.
+	Store LeaseStore
+	// Addr is the listen address (default ":8090"; --discovery-addr).
+	Addr string
+	// Token is the static bearer token from DISCOVERY_TOKEN. Empty
+	// disables auth — allowed for the prototype, but loudly warned about
+	// at startup, because in-cluster that is a silent security hole.
+	Token string
+	// MaxLeaseTTL caps requested lease TTLs (default 1h; --max-lease-ttl).
+	MaxLeaseTTL time.Duration
+
+	log logr.Logger
+
+	mu        sync.Mutex
+	boundAddr string
+}
+
+// NeedLeaderElection is false, and the deployment ships replicas: 1.
+// Verified against controller-runtime's runnable ordering: caches start and
+// sync before non-leader-election runnables, so cache reads here are safe.
+// If this were leader-elected, non-leader replicas would refuse connections
+// while still being Service endpoints. The 1-replica constraint comes from
+// lease state being per-process — both facts are README caveats.
+func (s *Server) NeedLeaderElection() bool { return false }
+
+// BoundAddr returns the actual listen address once Start has bound it —
+// meaningful when Addr uses port 0 (tests).
+func (s *Server) BoundAddr() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.boundAddr
+}
+
+// Start listens and serves until ctx ends, then shuts down gracefully with
+// a 10-second grace period.
+func (s *Server) Start(ctx context.Context) error {
+	if s.Addr == "" {
+		s.Addr = defaultAddr
+	}
+	if s.MaxLeaseTTL == 0 {
+		s.MaxLeaseTTL = defaultMaxTTL
+	}
+	s.log = logf.FromContext(ctx).WithName("discovery")
+	if s.Token == "" {
+		s.log.Info("WARNING: DISCOVERY_TOKEN is empty — the discovery API is served without authentication")
+	}
+
+	ln, err := net.Listen("tcp", s.Addr)
+	if err != nil {
+		return err
+	}
+	s.mu.Lock()
+	s.boundAddr = ln.Addr().String()
+	s.mu.Unlock()
+
+	srv := &http.Server{
+		Handler:           s.handler(),
+		ReadHeaderTimeout: readHeadTimeout,
+		ReadTimeout:       10 * time.Second,
+		WriteTimeout:      10 * time.Second,
+		IdleTimeout:       60 * time.Second,
+	}
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- srv.Serve(ln) }()
+	s.log.Info("discovery API listening", "addr", s.boundAddr)
+
+	select {
+	case err := <-errCh:
+		return err
+	case <-ctx.Done():
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGrace)
+		defer cancel()
+		if err := srv.Shutdown(shutdownCtx); err != nil {
+			return err
+		}
+		<-errCh // always http.ErrServerClosed after a clean Shutdown
+		return nil
+	}
+}
+
+// handler assembles the mux and the middleware chain, outermost first:
+// recover → request-log → body-size cap → bearer auth.
+func (s *Server) handler() http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok\n"))
+	})
+	mux.HandleFunc("GET /v1/proxies", s.handleListProxies)
+	mux.HandleFunc("POST /v1/leases", s.handleAcquireLease)
+	mux.HandleFunc("DELETE /v1/leases/{id}", s.handleReleaseLease)
+	mux.HandleFunc("POST /v1/leases/{id}/report", s.handleReportLease)
+
+	var h http.Handler = mux
+	h = s.authMiddleware(h)
+	h = maxBytesMiddleware(h)
+	h = s.logMiddleware(h)
+	h = s.recoverMiddleware(h)
+	return h
+}
+
+func (s *Server) recoverMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer func() {
+			if p := recover(); p != nil {
+				s.log.Error(nil, "panic in discovery handler", "panic", p, "path", r.URL.Path)
+				writeError(w, http.StatusInternalServerError, "internal", "internal server error")
+			}
+		}()
+		next.ServeHTTP(w, r)
+	})
+}
+
+// statusRecorder captures the response code for the request log.
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+}
+
+func (r *statusRecorder) WriteHeader(code int) {
+	r.status = code
+	r.ResponseWriter.WriteHeader(code)
+}
+
+func (s *Server) logMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/healthz" {
+			next.ServeHTTP(w, r) // probes are noise
+			return
+		}
+		rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
+		start := time.Now()
+		next.ServeHTTP(rec, r)
+		s.log.Info("request",
+			"method", r.Method, "path", r.URL.Path,
+			"status", rec.status, "duration", time.Since(start).String())
+	})
+}
+
+func maxBytesMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
+		next.ServeHTTP(w, r)
+	})
+}
+
+func (s *Server) authMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if s.Token == "" || r.URL.Path == "/healthz" {
+			next.ServeHTTP(w, r)
+			return
+		}
+		token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ")
+		if !ok || subtle.ConstantTimeCompare([]byte(token), []byte(s.Token)) != 1 {
+			writeError(w, http.StatusUnauthorized, "unauthorized", "missing or invalid bearer token")
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
+// errorBody is the shared error shape:
+// {"error":"<machine_code>","message":"<human>"}.
+type errorBody struct {
+	Error   string `json:"error"`
+	Message string `json:"message"`
+}
+
+func writeError(w http.ResponseWriter, status int, code, message string) {
+	writeJSON(w, status, errorBody{Error: code, Message: message})
+}
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(v)
+}
--- a/internal/discovery/server_test.go
+++ b/internal/discovery/server_test.go
@@ -0,0 +1,386 @@
+package discovery
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease"
+)
+
+func testProxy(name string, attrs map[string]string, healthy bool, mut ...func(*crawlv1alpha1.Proxy)) *crawlv1alpha1.Proxy {
+	p := &crawlv1alpha1.Proxy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name, Namespace: "default", UID: types.UID("uid-" + name),
+		},
+		Spec: crawlv1alpha1.ProxySpec{
+			Mode:       crawlv1alpha1.ModeExternal,
+			Endpoint:   &crawlv1alpha1.EndpointSpec{Host: "10.0.0.1", Port: 3128},
+			Attributes: attrs,
+		},
+	}
+	p.Status.IP = "10.0.0.1"
+	p.Status.Phase = crawlv1alpha1.PhaseReady
+	status := metav1.ConditionFalse
+	if healthy {
+		status = metav1.ConditionTrue
+	}
+	p.Status.Conditions = []metav1.Condition{{
+		Type: crawlv1alpha1.ConditionHealthy, Status: status,
+		Reason: "Probing", LastTransitionTime: metav1.Now(),
+	}}
+	for _, m := range mut {
+		m(p)
+	}
+	return p
+}
+
+func withMaxLeases(n int32) func(*crawlv1alpha1.Proxy) {
+	return func(p *crawlv1alpha1.Proxy) { p.Spec.MaxLeases = &n }
+}
+
+func withLatency(ms int64) func(*crawlv1alpha1.Proxy) {
+	return func(p *crawlv1alpha1.Proxy) { p.Status.LatencyMillis = ms }
+}
+
+// newTestServer wires the handler chain to a fake cache reader and a real
+// lease store, served over httptest.
+func newTestServer(t *testing.T, token string, proxies ...*crawlv1alpha1.Proxy) (*httptest.Server, *Server) {
+	t.Helper()
+	s := runtime.NewScheme()
+	if err := crawlv1alpha1.AddToScheme(s); err != nil {
+		t.Fatalf("scheme: %v", err)
+	}
+	builder := fake.NewClientBuilder().WithScheme(s)
+	for _, p := range proxies {
+		builder = builder.WithObjects(p)
+	}
+	srv := &Server{
+		Reader:      builder.Build(),
+		Store:       lease.NewStore(15 * time.Minute),
+		Token:       token,
+		MaxLeaseTTL: time.Hour,
+	}
+	ts := httptest.NewServer(srv.handler())
+	t.Cleanup(ts.Close)
+	return ts, srv
+}
+
+type response struct {
+	status int
+	body   map[string]any
+}
+
+func do(t *testing.T, ts *httptest.Server, method, path, token string, body any) response {
+	t.Helper()
+	var reader io.Reader
+	if body != nil {
+		if s, ok := body.(string); ok {
+			reader = bytes.NewBufferString(s)
+		} else {
+			b, err := json.Marshal(body)
+			if err != nil {
+				t.Fatalf("marshaling request body: %v", err)
+			}
+			reader = bytes.NewBuffer(b)
+		}
+	}
+	req, err := http.NewRequestWithContext(context.Background(), method, ts.URL+path, reader)
+	if err != nil {
+		t.Fatalf("building request: %v", err)
+	}
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	resp, err := ts.Client().Do(req)
+	if err != nil {
+		t.Fatalf("%s %s: %v", method, path, err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	out := response{status: resp.StatusCode}
+	raw, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("reading response: %v", err)
+	}
+	if len(raw) > 0 && resp.Header.Get("Content-Type") == "application/json" {
+		if err := json.Unmarshal(raw, &out.body); err != nil {
+			t.Fatalf("decoding response %q: %v", raw, err)
+		}
+	}
+	return out
+}
+
+func TestAuth(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "sekrit", testProxy("p1", nil, true))
+
+	if got := do(t, ts, http.MethodGet, "/v1/proxies", "", nil); got.status != http.StatusUnauthorized {
+		t.Errorf("no token: status %d, want 401", got.status)
+	}
+	if got := do(t, ts, http.MethodGet, "/v1/proxies", "wrong", nil); got.status != http.StatusUnauthorized {
+		t.Errorf("wrong token: status %d, want 401", got.status)
+	}
+	if got := do(t, ts, http.MethodGet, "/v1/proxies", "sekrit", nil); got.status != http.StatusOK {
+		t.Errorf("correct token: status %d, want 200", got.status)
+	}
+	if got := do(t, ts, http.MethodGet, "/healthz", "", nil); got.status != http.StatusOK {
+		t.Errorf("healthz without token: status %d, want 200 (always unauthenticated)", got.status)
+	}
+}
+
+func TestAuth_disabledWithEmptyToken(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "", testProxy("p1", nil, true))
+	if got := do(t, ts, http.MethodGet, "/v1/proxies", "", nil); got.status != http.StatusOK {
+		t.Errorf("status %d, want 200 with auth disabled", got.status)
+	}
+}
+
+func TestListProxies(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "",
+		testProxy("eu-healthy", map[string]string{"geo": "eu", "purpose": "crawl"}, true),
+		testProxy("eu-sick", map[string]string{"geo": "eu"}, false),
+		testProxy("us-healthy", map[string]string{"geo": "us"}, true),
+	)
+
+	tests := []struct {
+		name      string
+		query     string
+		wantCount int
+		wantFirst string
+	}{
+		{name: "no filter returns everything", query: "", wantCount: 3, wantFirst: "default/eu-healthy"},
+		{name: "healthy filter", query: "?healthy=true", wantCount: 2},
+		{name: "unhealthy filter", query: "?healthy=false", wantCount: 1, wantFirst: "default/eu-sick"},
+		{name: "attribute filter", query: "?attr.geo=eu", wantCount: 2},
+		{name: "attribute and health combined", query: "?attr.geo=eu&healthy=true", wantCount: 1, wantFirst: "default/eu-healthy"},
+		{name: "two attributes must both match", query: "?attr.geo=eu&attr.purpose=crawl", wantCount: 1, wantFirst: "default/eu-healthy"},
+		{name: "no matches is 200 with count 0", query: "?attr.geo=mars", wantCount: 0},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got := do(t, ts, http.MethodGet, "/v1/proxies"+tc.query, "", nil)
+			if got.status != http.StatusOK {
+				t.Fatalf("status %d, want 200", got.status)
+			}
+			count := int(got.body["count"].(float64))
+			proxies := got.body["proxies"].([]any)
+			if count != tc.wantCount || len(proxies) != tc.wantCount {
+				t.Fatalf("count = %d (len %d), want %d", count, len(proxies), tc.wantCount)
+			}
+			if tc.wantFirst != "" {
+				first := proxies[0].(map[string]any)
+				if first["id"] != tc.wantFirst {
+					t.Errorf("first id = %v, want %s", first["id"], tc.wantFirst)
+				}
+			}
+		})
+	}
+
+	t.Run("invalid healthy value is 400", func(t *testing.T) {
+		t.Parallel()
+		if got := do(t, ts, http.MethodGet, "/v1/proxies?healthy=maybe", "", nil); got.status != http.StatusBadRequest {
+			t.Errorf("status %d, want 400", got.status)
+		}
+	})
+}
+
+func TestAcquireLease_grantShape(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "",
+		testProxy("eu1", map[string]string{"geo": "eu"}, true, withLatency(30)),
+		testProxy("eu2", map[string]string{"geo": "eu"}, true, withLatency(10)),
+	)
+
+	got := do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{
+		"selector": map[string]string{"geo": "eu"},
+	})
+	if got.status != http.StatusCreated {
+		t.Fatalf("status %d (%v), want 201", got.status, got.body)
+	}
+	if got.body["leaseID"] == "" || got.body["leaseID"] == nil {
+		t.Error("empty leaseID")
+	}
+	if got.body["ttlSeconds"].(float64) != 300 {
+		t.Errorf("ttlSeconds = %v, want the 300 default", got.body["ttlSeconds"])
+	}
+	proxy := got.body["proxy"].(map[string]any)
+	if proxy["id"] != "default/eu2" {
+		t.Errorf("granted %v, want default/eu2 (lower latency at equal load)", proxy["id"])
+	}
+	if proxy["activeLeases"].(float64) != 1 {
+		t.Errorf("activeLeases = %v, want 1 (this grant included)", proxy["activeLeases"])
+	}
+	if _, err := time.Parse(time.RFC3339, got.body["expiresAt"].(string)); err != nil {
+		t.Errorf("expiresAt %v is not RFC3339: %v", got.body["expiresAt"], err)
+	}
+}
+
+func TestAcquireLease_noMatchBody(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "",
+		testProxy("eu-tiny", map[string]string{"geo": "eu"}, true, withMaxLeases(1)),
+		testProxy("eu-sick", map[string]string{"geo": "eu"}, false),
+	)
+
+	body := map[string]any{"selector": map[string]string{"geo": "eu"}}
+	if got := do(t, ts, http.MethodPost, "/v1/leases", "", body); got.status != http.StatusCreated {
+		t.Fatalf("first acquire: status %d, want 201", got.status)
+	}
+
+	got := do(t, ts, http.MethodPost, "/v1/leases", "", body)
+	if got.status != http.StatusConflict {
+		t.Fatalf("second acquire: status %d, want 409", got.status)
+	}
+	want := map[string]float64{"considered": 2, "atCapacity": 1, "inCooldown": 0, "unhealthy": 1}
+	for k, v := range want {
+		if got.body[k].(float64) != v {
+			t.Errorf("%s = %v, want %v (body %v)", k, got.body[k], v, got.body)
+		}
+	}
+	if got.body["error"] != "no_match" {
+		t.Errorf("error = %v, want no_match", got.body["error"])
+	}
+}
+
+func TestAcquireLease_badRequests(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "", testProxy("p1", nil, true))
+
+	tests := []struct {
+		name     string
+		body     any
+		wantCode string
+	}{
+		{name: "ttl above the cap", body: map[string]any{"ttlSeconds": 999999}, wantCode: "invalid_ttl"},
+		{name: "negative ttl", body: map[string]any{"ttlSeconds": -5}, wantCode: "invalid_ttl"},
+		{name: "malformed json", body: "{not json", wantCode: "invalid_body"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got := do(t, ts, http.MethodPost, "/v1/leases", "", tc.body)
+			if got.status != http.StatusBadRequest || got.body["error"] != tc.wantCode {
+				t.Errorf("= %d/%v, want 400/%s", got.status, got.body["error"], tc.wantCode)
+			}
+		})
+	}
+}
+
+func TestReleaseLease_alwaysNoContent(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "", testProxy("p1", nil, true))
+
+	got := do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{})
+	if got.status != http.StatusCreated {
+		t.Fatalf("acquire: status %d, want 201", got.status)
+	}
+	id := got.body["leaseID"].(string)
+
+	for _, path := range []string{"/v1/leases/" + id, "/v1/leases/" + id, "/v1/leases/never-existed"} {
+		if got := do(t, ts, http.MethodDelete, path, "", nil); got.status != http.StatusNoContent {
+			t.Errorf("DELETE %s: status %d, want 204", path, got.status)
+		}
+	}
+}
+
+func TestReportLease(t *testing.T) {
+	t.Parallel()
+	ts, _ := newTestServer(t, "", testProxy("p1", map[string]string{"geo": "eu"}, true))
+
+	got := do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{
+		"selector": map[string]string{"geo": "eu"}, "target": "example.com",
+	})
+	if got.status != http.StatusCreated {
+		t.Fatalf("acquire: status %d, want 201", got.status)
+	}
+	id := got.body["leaseID"].(string)
+	reportPath := fmt.Sprintf("/v1/leases/%s/report", id)
+
+	if got := do(t, ts, http.MethodPost, reportPath, "", map[string]any{"result": "rate_limited", "target": "example.com"}); got.status != http.StatusNoContent {
+		t.Fatalf("report: status %d, want 204", got.status)
+	}
+
+	// The cooldown from the report now blocks same-target acquisition.
+	got = do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{
+		"selector": map[string]string{"geo": "eu"}, "target": "example.com",
+	})
+	if got.status != http.StatusConflict || got.body["inCooldown"].(float64) != 1 {
+		t.Errorf("post-report acquire = %d/%v, want 409 with inCooldown 1", got.status, got.body)
+	}
+
+	t.Run("invalid result value", func(t *testing.T) {
+		got := do(t, ts, http.MethodPost, reportPath, "", map[string]any{"result": "throttled"})
+		if got.status != http.StatusBadRequest || got.body["error"] != "invalid_result" {
+			t.Errorf("= %d/%v, want 400/invalid_result", got.status, got.body["error"])
+		}
+	})
+	t.Run("unknown lease", func(t *testing.T) {
+		got := do(t, ts, http.MethodPost, "/v1/leases/never-existed/report", "", map[string]any{"result": "ok"})
+		if got.status != http.StatusNotFound || got.body["error"] != "unknown_lease" {
+			t.Errorf("= %d/%v, want 404/unknown_lease", got.status, got.body["error"])
+		}
+	})
+}
+
+func TestStart_servesAndShutsDown(t *testing.T) {
+	t.Parallel()
+	s := runtime.NewScheme()
+	if err := crawlv1alpha1.AddToScheme(s); err != nil {
+		t.Fatalf("scheme: %v", err)
+	}
+	srv := &Server{
+		Reader: fake.NewClientBuilder().WithScheme(s).Build(),
+		Store:  lease.NewStore(time.Minute),
+		Addr:   "127.0.0.1:0",
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- srv.Start(ctx) }()
+
+	var addr string
+	deadline := time.After(5 * time.Second)
+	for addr == "" {
+		select {
+		case <-deadline:
+			t.Fatal("server never bound")
+		case <-time.After(5 * time.Millisecond):
+			addr = srv.BoundAddr()
+		}
+	}
+
+	resp, err := http.Get("http://" + addr + "/healthz")
+	if err != nil {
+		t.Fatalf("healthz: %v", err)
+	}
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("healthz status %d, want 200", resp.StatusCode)
+	}
+
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("Start returned %v, want nil after graceful shutdown", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Start did not stop on cancel")
+	}
+}