From 9ba04b16e0263c2b2d09fc9be4f36ab53944d2e1 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Fri, 25 Sep 2026 08:28:07 -0700 Subject: [PATCH] v1.1 plan: cancelled requests recorded as 499, empty usage is []; tests proven failing on master Co-Authored-By: Claude Fable 5.1 --- docs/plans/v1.1/01-review-fixes.md | 66 +++++++++++ docs/plans/v1.1/README.md | 27 +++++ .../_files/internal/admin/usage_empty_test.go | 31 +++++ .../v1.1/_files/internal/proxy/cancel_test.go | 109 ++++++++++++++++++ 4 files changed, 233 insertions(+) create mode 100644 docs/plans/v1.1/01-review-fixes.md create mode 100644 docs/plans/v1.1/README.md create mode 100644 docs/plans/v1.1/_files/internal/admin/usage_empty_test.go create mode 100644 docs/plans/v1.1/_files/internal/proxy/cancel_test.go diff --git a/docs/plans/v1.1/01-review-fixes.md b/docs/plans/v1.1/01-review-fixes.md new file mode 100644 index 0000000..a9bab18 --- /dev/null +++ b/docs/plans/v1.1/01-review-fixes.md @@ -0,0 +1,66 @@ +# v1.1 task 01: review fixes — cancelled clients are recorded; empty usage is `[]` + +**Branch:** `v1.1` (create it from `master`: `git switch master && git switch -c v1.1`; `git status --short` must be empty first, otherwise stop) +**Commit subject:** `Review fixes: record cancelled requests as 499; empty usage is an array` + +## What the reviewer observed + +1. A client that disconnects mid-stream leaves **no accounting row**: after `curl -m 0.4 -N …` + against a streaming completion, `/_crossbar/usage` stayed empty. Task 05's rule 5 said the + reverse proxy's `ErrorHandler` does nothing on `context.Canceled`; rule 6 said "record what + you have when `ServeHTTP` returns". The second rule was not applied on that path, and the + same gap exists for a client that gives up while waiting in the limiter queue (rule 4 said + "just return, log 499"). Cancelled requests held a slot and cost prefill; usage and error + rate must see them. The task text was ambiguous (owner's fault); the fix is still needed. +2. `GET /_crossbar/usage` with no rows answers `null`. The spec said a JSON array. Clients iterate + the result; `null` is not iterable. + +## Files + +- Copy (never edit afterwards): `internal/proxy/cancel_test.go`, `internal/admin/usage_empty_test.go` +- Modify: files under `internal/proxy/` as needed (`forward.go`, `proxy.go`), `internal/admin/admin_ops.go` (or wherever the usage handler lives), `docs/implementer-log.md` + +## Rules + +1. **Every request that reached step 3 of `ServeHTTP` (a lease was acquired) writes exactly one + `store.Request` row**, on every exit path: normal completion, upstream error (502), queue full + (503), client cancelled while queued (**499**, `Err: "client cancelled while queued"`), client + cancelled during the forward (**499**, `Err: "client cancelled"`, with whatever tokens the tee + had seen). Detect the forward case with `r.Context().Err() != nil` after `rp.ServeHTTP` + returns, or in the `ErrorHandler` when `errors.Is(err, context.Canceled)`; do not write to the + client in that case, do not mark the host down, but do record. Status 499 is not an HTTP + status the client sees; it is the row's status (and the log line's), as nginx does. +2. **`/_crossbar/usage` JSON** encodes an empty result as `[]`: initialise the slice + (`rows := []store.UsageRow{}` / `make(..., 0)`) before encoding, on every `by` value and with + or without `since`. The text form prints its header line even with no rows. +3. Nothing else changes. Existing tests must keep passing; the two new ones must pass. + +## Steps + +- [ ] **1. Branch and copy.** + +```sh +git switch master && git switch -c v1.1 +cp docs/plans/v1.1/_files/internal/proxy/cancel_test.go internal/proxy/ +cp docs/plans/v1.1/_files/internal/admin/usage_empty_test.go internal/admin/ +``` + +- [ ] **2. See them fail.** `go test -run 'Cancel|UsageEmpty' ./internal/proxy/ ./internal/admin/`. + Expected: all three tests fail (`no 499 row`, `body "null"`). If one passes already, stop and report. +- [ ] **3. Fix.** `gofmt -w internal/`. +- [ ] **4. See everything pass.** `go test -race -count=2 ./...`. The cancel tests are timing-based with generous margins. +- [ ] **5. Run the gate.** `make gate`. Expected last line: `gate: ok`. +- [ ] **6. Log and commit.** Row `v1.1/01-review-fixes`. + +```sh +git add internal/proxy internal/admin docs/implementer-log.md +git commit +``` + +## Done when + +- Step 2 failed before the fix and `go test -race -count=2 ./...` passes after; `make gate` prints `gate: ok`; both copied tests byte-identical to `_files/`. + +## Stop and report if + +- Step 2 passes before any change, or the cancel tests fail intermittently after the fix (report the failure text). diff --git a/docs/plans/v1.1/README.md b/docs/plans/v1.1/README.md new file mode 100644 index 0000000..8ceb439 --- /dev/null +++ b/docs/plans/v1.1/README.md @@ -0,0 +1,27 @@ +# v1.1 implementation plan: review follow-ups + +> **For the implementing model:** do not work from this file. The owner gives you one task file at +> a time. This file is the index for the owner and the reviewer. + +**Goal:** close findings 1 and 2 of the v1 review (`docs/implementer-log.md`): a request whose +client disconnects — mid-stream or while queued — must still write its accounting row (status +499), and `/_crossbar/usage` with no rows must answer `[]`, not `null`. + +**How this plan was made:** acceptance tests first, from the findings; no reference +implementation. Both given tests were run against `master` at the merge of `v1`: all three fail +there (the two cancel tests find no 499 row; the empty-usage test gets `null`). + +## Tasks + +| # | File | Delivers | Tests that define it | +|---|---|---|---| +| 01 | `01-review-fixes.md` | 499 rows on both cancel paths; `[]` for empty usage | `internal/proxy/cancel_test.go`, `internal/admin/usage_empty_test.go` | + +Branch `v1.1`. One task, one fresh OpenCode session, one commit. + +## For the reviewer + +1. `git log --oneline master..v1.1`: one commit with the trailer. +2. `cmp` both copied tests; `git diff master..v1.1 --stat -- PLAN.md AGENTS.md docs/plans` empty. +3. `make gate`, `make smoke`. +4. Probe: cut a stream with `curl -m 0.4 -N …` against the smoke rig and confirm one `status="499"` line in `/_crossbar/metrics`. diff --git a/docs/plans/v1.1/_files/internal/admin/usage_empty_test.go b/docs/plans/v1.1/_files/internal/admin/usage_empty_test.go new file mode 100644 index 0000000..dc586f1 --- /dev/null +++ b/docs/plans/v1.1/_files/internal/admin/usage_empty_test.go @@ -0,0 +1,31 @@ +package admin_test + +import ( + "encoding/json" + "strings" + "testing" + + "git.wntrmute.dev/kyle/crossbar/internal/store" +) + +// An empty usage table is an empty JSON array, not null: clients iterate it. +func TestUsageEmptyIsAnArray(t *testing.T) { + r := newRig(t) + for _, q := range []string{"/_crossbar/usage", "/_crossbar/usage?by=host", "/_crossbar/usage?by=model&since=1h"} { + rec := r.do(t, "GET", q, "") + if rec.Code != 200 { + t.Fatalf("%s: %d", q, rec.Code) + } + if strings.TrimSpace(rec.Body.String()) != "[]" { + t.Errorf("%s: body %q, want []", q, rec.Body.String()) + } + var rows []store.UsageRow + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil || rows == nil || len(rows) != 0 { + t.Errorf("%s: decoded %v %v, want an empty non-nil slice", q, rows, err) + } + } + rec := r.do(t, "GET", "/_crossbar/usage?by=route", "", "Accept", "text/plain") + if rec.Code != 200 || !strings.Contains(rec.Body.String(), "key") { + t.Errorf("text form with no rows must still print the header: %d %q", rec.Code, rec.Body.String()) + } +} diff --git a/docs/plans/v1.1/_files/internal/proxy/cancel_test.go b/docs/plans/v1.1/_files/internal/proxy/cancel_test.go new file mode 100644 index 0000000..fba808b --- /dev/null +++ b/docs/plans/v1.1/_files/internal/proxy/cancel_test.go @@ -0,0 +1,109 @@ +package proxy_test + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "git.wntrmute.dev/kyle/crossbar/internal/store" +) + +// A client that goes away mid-stream is still a request that happened: it held a slot, it cost +// prefill, and it belongs in the accounting. The row records status 499 and a non-empty err. +func TestClientCancelMidStreamIsRecorded(t *testing.T) { + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/health": + fmt.Fprint(w, `{"status":"ok"}`) + case "/v1/models": + fmt.Fprint(w, `{"object":"list","data":[{"id":"shared"}]}`) + default: + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(200) + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"first\"}}]}\n\n") + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + case <-time.After(3 * time.Second): + } + } + })) + t.Cleanup(slow.Close) + beta := newUpstream(t, "beta") + r := newRig(t, twoHosts, &upstream{name: "alpha", srv: slow}, beta) + + ctx, cancel := context.WithCancel(context.Background()) + body := `{"model":"alpha-only","stream":true,"messages":[{"role":"user","content":"cancel me"}]}` + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, r.front.URL+"/r/v1/chat/completions", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + buf := make([]byte, 64) + if _, err := resp.Body.Read(buf); err != nil { + t.Fatalf("first chunk: %v", err) + } + cancel() + resp.Body.Close() + + deadline := time.Now().Add(3 * time.Second) + var counts []store.StatusCount + for time.Now().Before(deadline) { + counts, _ = r.store.StatusCounts(time.Time{}) + if len(counts) > 0 { + break + } + time.Sleep(25 * time.Millisecond) + } + if len(counts) != 1 || counts[0].Status != 499 || counts[0].Route != "r" || counts[0].Count != 1 { + t.Fatalf("status counts after a cancelled stream = %+v, want one row: route r, status 499", counts) + } + rows, _ := r.store.Usage(time.Time{}, store.ByRoute) + if len(rows) != 1 || rows[0].Requests != 1 || rows[0].Errors != 1 { + t.Errorf("usage = %+v, want 1 request counted as an error", rows) + } +} + +// The same when the client gives up while waiting in the queue: a 499 row, no slot leaked. +func TestClientCancelWhileQueuedIsRecorded(t *testing.T) { + alpha := newUpstream(t, "alpha") + alpha.delay = 800 * time.Millisecond + r := newRig(t, ` +listen = "127.0.0.1:1" +queue_max = 2 +[hosts.alpha] +base_url = %q +models = { "shared" = { parallel = 1 } } +[routes.r] +hosts = ["alpha"] +default_model = "shared" +`, alpha) + go func() { drain(r.post("/r/v1/chat/completions", conversation(1, 1))) }() // holds the one slot + time.Sleep(100 * time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, r.front.URL+"/r/v1/chat/completions", strings.NewReader(conversation(2, 1))) + req.Header.Set("Content-Type", "application/json") + if _, err := http.DefaultClient.Do(req); err == nil { + t.Fatal("the queued request should have been cancelled by its context") + } + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + counts, _ := r.store.StatusCounts(time.Time{}) + for _, c := range counts { + if c.Status == 499 { + if r.lim.Queued("alpha", "shared") != 0 { + t.Errorf("queued = %d after the waiter cancelled", r.lim.Queued("alpha", "shared")) + } + return + } + } + time.Sleep(25 * time.Millisecond) + } + t.Fatal("no 499 row recorded for the request cancelled while queued") +}