crossbar/docs/plans/v1/05-proxy.md

v1 task 05: the proxy uses leases, the limiter, the SSE tee and the store

Branch: v1 (run git switch v1; git status --short must be empty, otherwise stop) Commit subject: Route by lease, queue per host and model, record every request

Goal

internal/proxy becomes the v1 proxy: route from path or header, fingerprint the body, acquire a lease, take a limiter slot (queue or 503), forward with streaming, tee the response to read usage/timings, mark hosts down on failure, and record one store.Request per request. PLAN.md §4, §6, §7a.

Context

The v0 proxy stays the skeleton of this one: SplitRoute, the ordered error answers, the model peek, httputil.ReverseProxy with FlushInterval: -1, the status recorder with a checked Flush, the log line. What changes is who picks the host and what happens around the forward. Two new response headers make the behaviour observable: X-Crossbar-Host (existing) and X-Crossbar-Lease: new|reused. The given test drives the whole handler over real HTTP with a real store, lease.Table, limiter and health.Table.

Files

• Copy (replaces v0's file): internal/proxy/proxy_test.go
• Copy: internal/proxy/helpers_test.go (the fakeHealth helper that v0's proxy_test.go held and recorder_test.go still needs)
• Modify: internal/proxy/proxy.go (split into more files if it passes 400 lines: proxy.go, tee.go, hosts.go), docs/implementer-log.md
• Keep: internal/proxy/recorder_test.go from v0.1 — it must still pass. Its proxy.New(cfg, h, nil) call no longer compiles, so this is the one given test you edit: change that call to proxy.New(cfg, h, nil, nil, nil, nil) and nothing else; New must accept nils for leases, lim, rec and then behave like v0 (first healthy host, no queue, no recording). Say so in Deviations.

Interfaces

internal/proxy, package proxy (v0 names kept; additions):

const (
    MaxBody     = 16 << 20
    HostHeader  = "X-Crossbar-Host"
    LeaseHeader = "X-Crossbar-Lease"   // "new" or "reused"
    RouteHeader = "X-Crossbar-Route"   // client may name the route here instead of the path
)
type Health interface { Get(name string) (health.Status, bool); MarkDown(name, reason string) }
type Recorder interface { RecordRequest(store.Request) error }   // *store.Store satisfies it

// Hosts adapts the health table and config for the lease table, and holds the drain set.
type Hosts struct { /* private */ }
func HostView(h *health.Table, cfg *config.Config) *Hosts
func (h *Hosts) Healthy(name string) bool
func (h *Hosts) Draining(name string) bool
func (h *Hosts) SetDraining(name string, on bool)

// Chooser adapts config, health and limiter to lease.Chooser using choose.Best:
// Info{Healthy, Draining: false (the lease table already filtered), Loaded: model in Loaded,
// CanServe: cfg.Serves, Free: free slots FOR THIS MODEL on this host =
// cfg.Hosts[host].Models[model].Parallel - lim.InFlight(host, model) (never below 0; 0 when the
// host does not list the model), Queued: lim.Queued(host, model), Weight}.
// (Corrected 2026-09-25: an earlier version said lim.FreeSlots(host), which sums every model's
// slots and let a host win on slots the requested model cannot use.)
func Chooser(cfg *config.Config, h *health.Table, l *limiter.Limiter) lease.Chooser

func New(cfg *config.Config, h Health, leases *lease.Table, lim *limiter.Limiter, rec Recorder, log *slog.Logger) *Handler
func (p *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)
func SplitRoute(path string) (route, rest string, ok bool)

ServeHTTP, in order (every error answer is JSON {"error":"…"} as in v0):

1. Route. hdr := r.Header.Get(RouteHeader). If hdr != "": the path is used whole as rest (it must then start with /v1/ or be /health or /props); if the path also starts with a known route name and it differs from hdr → 400 conflicting route. If hdr == "": SplitRoute as in v0 (400 missing route). Unknown route (either source) → 404 unknown route. Disallowed rest → 404 not found.
2. Peek (v0 rule): body up to MaxBody → 413; model from the body or the route default; fp := fingerprint.Of(body) (GET/HEAD → "").
3. Lease. host, reused, err := leases.Acquire(lease.Key{route, fp, model}, rt.Hosts, time.Now()). ErrNoHost → 503 no healthy host; ErrPinnedDown → 503 pinned host down.
4. Slot. release, waited, err := lim.Acquire(r.Context(), host, model). ErrQueueFull → 503 queue full; ctx error → 499-style: just return (the client left; log status 499). defer release().
5. Forward as in v0 (Rewrite, FlushInterval: -1, ModifyResponse sets HostHeader and LeaseHeader, ErrorHandler marks down + 502 with host). Tee: in ModifyResponse, wrap resp.Body in a reader that passes every byte through unchanged and, when Content-Type starts with text/event-stream, scans complete data: lines for a JSON object with usage and/or timings, remembering the last one seen; for non-streamed JSON answers, remember the whole body's usage/timings (bounded: keep at most 1 MiB for the parse; beyond that, record no tokens). The scanner must not hold data back: Read returns what the upstream returned.
6. Record, after the upstream body is closed (the tee's Close, or the error handler): store.Request{Route, FP: fp, Model, Host, Started, QueuedMs: waited, TTFBMs (first byte of the response head), TotalMs, Status, Streamed, PromptTokens: usage.prompt_tokens (or timings.prompt_n), CachedTokens: timings.cache_n, CompletionTokens: usage.completion_tokens (or timings.predicted_n), Err}. Also record the 503/502 cases (Status set, no tokens). Do it from the request goroutine after rp.ServeHTTP returns, so tests that read the store right after the response see the row; if the tee cannot tell that the body closed, record what you have when ServeHTTP returns. A rec error is logged, never returned to the client.
7. When leases == nil (v0.1 compatibility path used by recorder_test.go): choose with the v0 Choose rule, skip the limiter and the store, still set HostHeader.
8. Log line as v0, adding lease=new|reused, queued_ms, fp (first 8 hex chars only). Never the body.

Steps

• 1. Copy (replace). git switch v1; cp docs/plans/v1/_files/internal/proxy/proxy_test.go internal/proxy/proxy_test.go; cp docs/plans/v1/_files/internal/proxy/helpers_test.go internal/proxy/. Edit the one call in internal/proxy/recorder_test.go as described above.
• 2. See it fail (compile). 3. Write the code. gofmt -w internal/proxy/.
• 4. See it pass. go test -race -count=1 ./internal/proxy/. TestDifferentConversationsSpreadByFreeSlots and TestQueueFullIs503 are timing-based with generous margins; run -count=3.
• 5. Run the gate. make gate. cmd/crossbar will not compile until task 06 — if go vet ./... fails only in cmd/crossbar/main.go because of the new New signature, change that one call to pass nil, nil, nil for the new arguments (task 06 wires it properly) and say so in Deviations.
• 6. Log and commit. Row v1/05-proxy.

git add internal/proxy cmd/crossbar docs/implementer-log.md
git commit

Done when

• go test -race -count=3 ./internal/proxy/ is ok; make gate prints gate: ok; cmp internal/proxy/proxy_test.go docs/plans/v1/_files/internal/proxy/proxy_test.go prints nothing.

Stop and report if

• TestAccountingRowsFromUsageAndTimings fails on the token sums while the stream test passes: quote the recorded row.