v1.1 plan: cancelled requests recorded as 499, empty usage is []; tests proven failing on master

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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).

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`.

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())
+	}
+}

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")
+}