From 999e1bf6909fa50a34433a37d686f1d2af5764aa Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Thu, 3 Sep 2026 13:06:20 +0300 Subject: [PATCH 1/2] test: synchronise every fixture-upstream capture a test goroutine reads A fixture handler runs on its httptest.Server's own goroutine. The HTTP round trip that follows LOOKS like it orders the handler's writes before the test goroutine's reads, but it is not a happens-before edge the memory model (or the race detector) recognises: a `var got []byte` written in the handler and read after `resp.Body.Close()` is an unsynchronised access, and `make cover` runs the suite with -race. Swept all 281 _test.go files: 95 closures run on another goroutine (handler bodies and `go func`), 46 write a variable the enclosing test reads, and 30 of those writes had no mutex, channel, WaitGroup or atomic between the two accesses. The other 16 were already correct -- tenancy_test.go's hostedFixture (f.mu), modes_test.go's captureUpstream (mu.Lock), the wg.Wait()-joined slice fills in diag_test.go and modes_test.go, and the atomic.Int64 counters expandsplice_test.go already uses. Three shapes, chosen so the edge is visible at the call site: - Bodies, headers and per-round captures in package proxy_test go through a new mutex-guarded `upstreamCapture` (proxy_test.go), modelled on hostedFixture. `record()` returns the 1-based round number, so a handler that answers differently per round no longer needs a captured counter either. - Counter-only fixtures become `atomic.Int64`, which carries its own edge and is the smaller change -- and the shape expandsplice_test.go already used. - Single-shot struct captures (cheapmodel, keepalive, keepalive_wire, tenancy) come back over a buffered channel; prefixask's fixture gets a `forwarded()` accessor behind its own mutex. Behaviour is unchanged: every assertion still reads the same bytes, and the two tests where "no request arrived" is itself the assertion (keepalive_wire, tenancy) use a non-blocking receive so that case stays observable instead of deadlocking. Honest note on verification: -race does NOT report this shape, before or after. A standalone reproducer of the canonical pattern is clean at -count=200 (including a variant that assigns after the response is flushed), while a control with two concurrent handlers writing one variable is reported immediately -- so the detector was live and the loopback path is manufacturing a happens-before edge through net/http's shared internals. These are therefore findings by inspection: unsynchronised by the memory model, not currently observable by TSan. proxy/conformance_test.go and proxy/counttokens_test.go are deliberately untouched; they are fixed separately in #141, which adds its own `recordedRequest` helper. The helper added here is named `upstreamCapture` so the two cannot collide at merge. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: DAVID AMID --- internal/cheapmodel/cheapmodel_test.go | 22 ++-- proxy/adjudicatetool_test.go | 31 +++-- proxy/agentcompaction_test.go | 13 +- proxy/dashexpand_test.go | 10 +- proxy/expandgate_test.go | 6 +- proxy/expandsplice_test.go | 26 ++-- proxy/keepalive_test.go | 9 +- proxy/keepalive_wire_test.go | 15 ++- proxy/prefixask_test.go | 25 +++- proxy/proxy_test.go | 166 +++++++++++++++++-------- proxy/tenancy_test.go | 11 +- 11 files changed, 216 insertions(+), 118 deletions(-) diff --git a/internal/cheapmodel/cheapmodel_test.go b/internal/cheapmodel/cheapmodel_test.go index 06df247d..62bd99c4 100644 --- a/internal/cheapmodel/cheapmodel_test.go +++ b/internal/cheapmodel/cheapmodel_test.go @@ -19,12 +19,17 @@ func TestAnthropicSkipsNonTextLeadingBlock(t *testing.T) { } } +// authHeaders is what a fixture upstream saw on the wire. Handlers run on the test server's +// own goroutine, so the values travel back over a buffered channel rather than through a +// captured variable: the HTTP round trip is not a happens-before edge, and a plain capture +// read by the test goroutine is a data race whether or not -race happens to observe it. +type authHeaders struct{ auth, key, version string } + func TestAnthropicBearerAuth(t *testing.T) { - var gotAuth, gotKey, gotVersion string + seen := make(chan authHeaders, 1) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotAuth = r.Header.Get("Authorization") - gotKey = r.Header.Get("x-api-key") - gotVersion = r.Header.Get("anthropic-version") + seen <- authHeaders{r.Header.Get("Authorization"), r.Header.Get("x-api-key"), + r.Header.Get("anthropic-version")} _, _ = io.WriteString(w, `{"content":[{"type":"text","text":"OK"}]}`) })) defer srv.Close() @@ -32,6 +37,8 @@ func TestAnthropicBearerAuth(t *testing.T) { if err != nil { t.Fatalf("err %v", err) } + got := <-seen + gotAuth, gotKey, gotVersion := got.auth, got.key, got.version if gotAuth != "Bearer tok" { t.Fatalf("Authorization = %q, want %q", gotAuth, "Bearer tok") } @@ -44,10 +51,9 @@ func TestAnthropicBearerAuth(t *testing.T) { } func TestAnthropicDefaultAuthUsesAPIKey(t *testing.T) { - var gotAuth, gotKey string + seen := make(chan authHeaders, 1) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotAuth = r.Header.Get("Authorization") - gotKey = r.Header.Get("x-api-key") + seen <- authHeaders{auth: r.Header.Get("Authorization"), key: r.Header.Get("x-api-key")} _, _ = io.WriteString(w, `{"content":[{"type":"text","text":"OK"}]}`) })) defer srv.Close() @@ -55,6 +61,8 @@ func TestAnthropicDefaultAuthUsesAPIKey(t *testing.T) { if err != nil { t.Fatalf("err %v", err) } + got := <-seen + gotAuth, gotKey := got.auth, got.key if gotKey != "tok" { t.Fatalf("x-api-key = %q, want %q", gotKey, "tok") } diff --git a/proxy/adjudicatetool_test.go b/proxy/adjudicatetool_test.go index af49ed1f..9b30b95a 100644 --- a/proxy/adjudicatetool_test.go +++ b/proxy/adjudicatetool_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "github.com/tidwall/gjson" @@ -32,9 +33,9 @@ func forwardedBody(t *testing.T, body []byte) []byte { // point is that one of those two differs. func forwardedOn(t *testing.T, route, yaml string, body []byte) []byte { t.Helper() - var got []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "application/json") if strings.Contains(route, "anthropic") { _, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` + @@ -53,7 +54,7 @@ func forwardedOn(t *testing.T, route, yaml string, body []byte) []byte { t.Fatal(err) } resp.Body.Close() - return got + return up.last().body } // toolsRequest is the ANTHROPIC dialect, because that is the only dialect the injection targets: @@ -210,11 +211,9 @@ func TestAdjudicateToolNotAdvertisedOnANonAnthropicRoute(t *testing.T) { // must instead be answered IN BAND, before the client is written to, which leaves the request-path // repair as a backstop rather than the primary defence. func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) { - round := 0 - var second []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - body, _ := io.ReadAll(r.Body) - round++ + round := up.record(r) w.Header().Set("Content-Type", "application/json") if round == 1 { // The model calls OUR tool, which is always a defect by construction. @@ -224,7 +223,6 @@ func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) { `"usage":{"input_tokens":5,"output_tokens":1}}`)) return } - second = body _, _ = w.Write([]byte(`{"id":"m2","type":"message","role":"assistant","model":"claude",` + `"content":[{"type":"text","text":"done"}],"stop_reason":"end_turn",` + `"usage":{"input_tokens":6,"output_tokens":2}}`)) @@ -242,10 +240,11 @@ func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) { resp.Body.Close() // PRECONDITION: the loop must actually have run a second round, or this asserts nothing. - if round < 2 { + if round := up.hits(); round < 2 { t.Fatalf("the stray was never intercepted -- only %d upstream round(s), client got: %s", round, got) } + second := up.body(2) if strings.Contains(string(got), adjudicate.ToolName) { t.Errorf("the proxy-injected tool_use reached the CLIENT: %s", got) } @@ -279,10 +278,10 @@ func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) { }} // --- Turn 1: the leak. ------------------------------------------------------------------- - rounds := 0 + var rounds atomic.Int64 // an atomic carries the happens-before edge the round trip does not upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.ReadAll(r.Body) - rounds++ + rounds.Add(1) w.Header().Set("Content-Type", "application/json") blocks, _ := json.Marshal(assistantCoCall["content"]) _, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` + @@ -303,7 +302,7 @@ func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) { // PRECONDITION: exactly one upstream round. A continuation would mean the loop answered in band // after all, and then the rest of this test is asserting nothing about the co-call path. - if rounds != 1 { + if rounds := rounds.Load(); rounds != 1 { t.Fatalf("expected the loop to bail on otherTools after ONE round, got %d — the co-call path "+ "no longer defers, so this test's premise is gone: %s", rounds, leaked) } @@ -362,11 +361,11 @@ func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) { // THE LEAK, streaming path. The splicer withheld only the expand tool by name, so an adjudication call // streamed through event by event and the client saw it live. func TestAdjudicateStrayCallDoesNotReachTheClientOnTheSSEPath(t *testing.T) { - round := 0 + var round atomic.Int64 // an atomic carries the happens-before edge the round trip does not upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.ReadAll(r.Body) - round++ - if round == 1 { + n := round.Add(1) + if n == 1 { w.Header().Set("Content-Type", "text/event-stream") w.WriteHeader(http.StatusOK) for _, ev := range []string{ @@ -415,7 +414,7 @@ func TestAdjudicateStrayCallDoesNotReachTheClientOnTheSSEPath(t *testing.T) { } got, _ := io.ReadAll(resp.Body) resp.Body.Close() - if round < 2 { + if round := round.Load(); round < 2 { t.Fatalf("the streamed stray was never intercepted -- only %d round(s), client got: %s", round, got) } diff --git a/proxy/agentcompaction_test.go b/proxy/agentcompaction_test.go index 7c586322..ca059da2 100644 --- a/proxy/agentcompaction_test.go +++ b/proxy/agentcompaction_test.go @@ -98,9 +98,9 @@ func TestAgentCompactionIsBypassed(t *testing.T) { }, } { t.Run(tc.name, func(t *testing.T) { - var got []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`)) })) @@ -126,6 +126,7 @@ func TestAgentCompactionIsBypassed(t *testing.T) { } resp.Body.Close() + got := up.last().body if tc.wantBypass { if !bytes.Equal(got, body) { t.Fatalf("a compaction request must be forwarded byte-identical.\n sent: %s\n got: %s", body, got) @@ -221,9 +222,9 @@ func TestExpandToolAdvertisedOnEveryTurnButNeverOnACompaction(t *testing.T) { []map[string]any{marked, {"role": "user", "content": ccCompactPrompt}}, false}, } { t.Run(tc.name, func(t *testing.T) { - var got []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`)) })) @@ -253,9 +254,9 @@ func TestExpandToolAdvertisedOnEveryTurnButNeverOnACompaction(t *testing.T) { } resp.Body.Close() - advertised := strings.Contains(string(got), "context_guru_expand") + advertised := strings.Contains(string(up.last().body), "context_guru_expand") if advertised != tc.want { - t.Fatalf("expand tool advertised=%v, want %v: %s", advertised, tc.want, got) + t.Fatalf("expand tool advertised=%v, want %v: %s", advertised, tc.want, up.last().body) } }) } diff --git a/proxy/dashexpand_test.go b/proxy/dashexpand_test.go index 9e230d72..140ba3d6 100644 --- a/proxy/dashexpand_test.go +++ b/proxy/dashexpand_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "github.com/rossoctl/context-guru/dash" @@ -15,12 +16,11 @@ import ( // tile read zero while /stats showed the true count, and Overview.SavedAdjusted // (SavedUnique − ExpandTokens) OVER-REPORTED net savings by the whole bounce. func TestDashboardRecordsExpands(t *testing.T) { - var calls int + var calls atomic.Int64 // an atomic carries the happens-before edge the round trip does not up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.Copy(io.Discard, r.Body) - calls++ w.Header().Set("Content-Type", "application/json") - if calls == 1 { + if calls.Add(1) == 1 { // The model asks for the offloaded original back. w.Write([]byte(`{"choices":[{"message":{"role":"assistant","tool_calls":[` + `{"id":"call_1","type":"function","function":{"name":"context_guru_expand","arguments":"{\"id\":\"HASH\"}"}}` + @@ -46,8 +46,8 @@ func TestDashboardRecordsExpands(t *testing.T) { t.Fatal(err) } resp.Body.Close() - if calls != 2 { - t.Fatalf("expected the expand continuation (2 upstream calls), got %d", calls) + if n := calls.Load(); n != 2 { + t.Fatalf("expected the expand continuation (2 upstream calls), got %d", n) } waitForRows(t, rec, 1) diff --git a/proxy/expandgate_test.go b/proxy/expandgate_test.go index 483a9b5e..359e5b18 100644 --- a/proxy/expandgate_test.go +++ b/proxy/expandgate_test.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "github.com/tidwall/gjson" - "io" "net/http" "net/http/httptest" "strings" @@ -59,9 +58,9 @@ func TestExpandToolIsAdvertisedOnlyWhereMarkersCanExist(t *testing.T) { {"a pipeline with an offloader", "pipeline: [linecap, cachesplit]\n", true}, } { t.Run(c.name, func(t *testing.T) { - var forwarded []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - forwarded, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, `{"type":"message","usage":{"input_tokens":1}}`) })) @@ -89,6 +88,7 @@ func TestExpandToolIsAdvertisedOnlyWhereMarkersCanExist(t *testing.T) { t.Fatal(err) } resp.Body.Close() + forwarded := up.last().body if len(forwarded) == 0 { t.Fatal("upstream received nothing") } diff --git a/proxy/expandsplice_test.go b/proxy/expandsplice_test.go index 36349555..8ecf6052 100644 --- a/proxy/expandsplice_test.go +++ b/proxy/expandsplice_test.go @@ -60,18 +60,15 @@ const round2Answer = "event: message_start\ndata: {\"type\":\"message_start\",\" func TestExpandCalledAfterALeadingBlockIsNeverGivenToTheClient(t *testing.T) { for _, leadType := range []string{"text", "thinking"} { t.Run(leadType, func(t *testing.T) { - var calls int - var secondBody []byte + var up upstreamCapture head, tail := leadThenExpand(leadType) upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - b, _ := io.ReadAll(r.Body) - calls++ + round := up.record(r) w.Header().Set("Content-Type", "text/event-stream") - if calls == 1 { + if round == 1 { w.Write([]byte(head + tail)) return } - secondBody = b w.Write([]byte(round2Answer)) })) defer upstream.Close() @@ -93,9 +90,10 @@ func TestExpandCalledAfterALeadingBlockIsNeverGivenToTheClient(t *testing.T) { t.Fatalf("the client received a raw tool_use for OUR tool — this is the "+ "reported bug (`No such tool available: context_guru_expand`):\n%s", out) } - if calls != 2 { + if calls := up.hits(); calls != 2 { t.Fatalf("the expand call must drive a continuation round, got %d upstream calls", calls) } + secondBody := up.body(2) if !strings.Contains(string(secondBody), "THE ORIGINAL CONTENT") { t.Fatalf("continuation must carry the resolved original: %s", secondBody) } @@ -202,9 +200,9 @@ func TestTheStreamedPrefixReachesTheClientBeforeTheExpandCall(t *testing.T) { // This is the whole sequence a user lives through, in two requests, because that is the only // place it is visible: the leak on the first and the repair on the second. func TestTheClientsNoSuchToolErrorNeverReachesTheModel(t *testing.T) { - var lastUpstream []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - lastUpstream, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "text/event-stream") w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"role\":\"assistant\"}}\n\n" + "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n" + @@ -260,6 +258,7 @@ func TestTheClientsNoSuchToolErrorNeverReachesTheModel(t *testing.T) { io.Copy(io.Discard, resp2.Body) resp2.Body.Close() + lastUpstream := up.last().body if strings.Contains(string(lastUpstream), "No such tool available") { t.Fatalf("the model received the client's error for OUR tool:\n%s", lastUpstream) } @@ -342,12 +341,11 @@ func TestAFailedContinuationRoundStillEndsTheClientsTurn(t *testing.T) { // client's turn must still terminate: it gets the events the splice withheld, which is a // complete message, rather than a stream cut off mid-block. func TestAJSONContinuationRoundStillEndsTheClientsTurn(t *testing.T) { - var calls int + var calls atomic.Int64 head, tail := leadThenExpand("text") upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.Copy(io.Discard, r.Body) - calls++ - if calls > 1 { + if calls.Add(1) > 1 { w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"type":"message","role":"assistant","content":[{"type":"text","text":"ANSWERED"}]}`)) return @@ -370,8 +368,8 @@ func TestAJSONContinuationRoundStillEndsTheClientsTurn(t *testing.T) { out, _ := io.ReadAll(resp.Body) resp.Body.Close() - if calls != 2 { - t.Fatalf("expected the continuation round, got %d upstream calls", calls) + if n := calls.Load(); n != 2 { + t.Fatalf("expected the continuation round, got %d upstream calls", n) } if strings.Contains(string(out), `{"type":"message","role":"assistant"`) { t.Fatalf("a JSON body was appended to an open event stream:\n%s", out) diff --git a/proxy/keepalive_test.go b/proxy/keepalive_test.go index f994c10c..2fd62e39 100644 --- a/proxy/keepalive_test.go +++ b/proxy/keepalive_test.go @@ -688,15 +688,19 @@ func TestKillSwitch(t *testing.T) { // End to end against a real HTTP server: the ping actually goes out, carries the caller's // credential, and reaches the upstream as a max_tokens:1 non-streaming POST of the same body. func TestSendPingHitsTheUpstream(t *testing.T) { - var got struct { + // The handler runs on the test server's goroutine; the HTTP round trip is not a + // happens-before edge, so what it saw comes back over a buffered channel rather than + // through a captured variable the test goroutine would read unsynchronised. + type seen struct { body []byte auth string path string } + seenCh := make(chan seen, 1) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { b := make([]byte, 1<<20) n, _ := r.Body.Read(b) - got.body, got.auth, got.path = b[:n], r.Header.Get("Authorization"), r.URL.Path + seenCh <- seen{b[:n], r.Header.Get("Authorization"), r.URL.Path} w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"usage":{"input_tokens":0,"output_tokens":1,` + `"cache_read_input_tokens":48576,"cache_creation_input_tokens":0},"stop_reason":"max_tokens"}`)) @@ -716,6 +720,7 @@ func TestSendPingHitsTheUpstream(t *testing.T) { if u.CacheRead != 48576 || u.CacheWrite != 0 { t.Errorf("usage read=%d write=%d; a ping must READ the prefix, not write it", u.CacheRead, u.CacheWrite) } + got := <-seenCh if got.auth != "Bearer sk-caller" { t.Errorf("upstream saw Authorization %q", got.auth) } diff --git a/proxy/keepalive_wire_test.go b/proxy/keepalive_wire_test.go index 94077e94..5dd7dd95 100644 --- a/proxy/keepalive_wire_test.go +++ b/proxy/keepalive_wire_test.go @@ -26,14 +26,20 @@ func TestPingWireBytesEndToEndThroughRealSend(t *testing.T) { const cred = "Bearer sk-caller-END-TO-END" const marker = "MY-PRIVATE-SOURCE-CODE-MARKER" - var got struct { + // The handler runs on the test server's goroutine, and the HTTP round trip is not a + // happens-before edge, so what it saw comes back over a buffered channel. k.dispatch is + // k.fire (inline), so by the time sweep() returns the ping has completed and the value is + // already queued -- hence a non-blocking receive, which keeps "no ping fired" observable + // as the zero value rather than turning it into a deadlock. + type seen struct { body []byte auth string } + seenCh := make(chan seen, 1) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { b := make([]byte, 1<<20) n, _ := r.Body.Read(b) - got.body, got.auth = append([]byte(nil), b[:n]...), r.Header.Get("Authorization") + seenCh <- seen{append([]byte(nil), b[:n]...), r.Header.Get("Authorization")} w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"usage":{"input_tokens":0,"output_tokens":1,` + `"cache_read_input_tokens":48576,"cache_creation_input_tokens":0}}`)) @@ -83,6 +89,11 @@ func TestPingWireBytesEndToEndThroughRealSend(t *testing.T) { if n := k.sweep(clock.advance(281 * time.Second)); n != 1 { t.Fatalf("sweep fired %d pings", n) } + var got seen + select { + case got = <-seenCh: + default: + } if got.body == nil { t.Fatal("the upstream received no ping") } diff --git a/proxy/prefixask_test.go b/proxy/prefixask_test.go index b00a6bab..ba68a9cf 100644 --- a/proxy/prefixask_test.go +++ b/proxy/prefixask_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" bschemas "github.com/maximhq/bifrost/core/schemas" @@ -23,17 +24,29 @@ import ( // capturePrefixed stands in for the provider and records the body it was sent, so a test can assert // what actually reached the wire rather than what the caller intended. type capturePrefixed struct { + mu sync.Mutex body []byte srv *httptest.Server } +// forwarded is what the fixture last received. The handler runs on the test server's own +// goroutine and the HTTP round trip is not a happens-before edge, so the capture goes +// through c.mu -- the shape tenancy_test.go's hostedFixture uses. +func (c *capturePrefixed) forwarded() []byte { + c.mu.Lock() + defer c.mu.Unlock() + return c.body +} + func newCapturePrefixed(t *testing.T, cacheRead int) *capturePrefixed { t.Helper() c := &capturePrefixed{} c.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { b := make([]byte, r.ContentLength) _, _ = r.Body.Read(b) + c.mu.Lock() c.body = b + c.mu.Unlock() w.Header().Set("content-type", "application/json") _, _ = w.Write([]byte(`{"content":[{"text":"[]"}],"usage":{"input_tokens":12,` + `"output_tokens":3,"cache_creation_input_tokens":0,"cache_read_input_tokens":` + @@ -74,7 +87,7 @@ func TestCompletePrefixedAppendsWithoutDisturbingThePrefix(t *testing.T) { t.Fatalf("CompletePrefixed: %v", err) } // PRECONDITION: the call reached the stub at all, or every assertion below is about nothing. - if len(srv.body) == 0 { + if len(srv.forwarded()) == 0 { t.Fatal("no body reached the provider") } if reply != "[]" { @@ -94,8 +107,8 @@ func TestCompletePrefixedAppendsWithoutDisturbingThePrefix(t *testing.T) { Content any `json:"content"` } `json:"messages"` } - if err := json.Unmarshal(srv.body, &sent); err != nil { - t.Fatalf("the body sent is not JSON: %v\n%s", err, srv.body) + if err := json.Unmarshal(srv.forwarded(), &sent); err != nil { + t.Fatalf("the body sent is not JSON: %v\n%s", err, srv.forwarded()) } // The ask is the LAST message and a USER one. The route rejects assistant prefill, which this // satisfies by construction — and it is why the prefix must not be extended any other way. @@ -218,7 +231,7 @@ func TestCompletePrefixedRefusesABodyWithNoMessages(t *testing.T) { if _, _, err := cli.CompletePrefixed(context.Background(), []byte(`{"model":"m"}`), "ask"); err == nil { t.Fatal("a body with no messages array was accepted") } - if len(srv.body) != 0 { + if len(srv.forwarded()) != 0 { t.Error("a malformed prefix was still sent to the provider") } } @@ -255,10 +268,10 @@ func TestAskUsesTheStashedBodyForThatSession(t *testing.T) { if u.CacheRead != 4242 { t.Fatalf("CacheRead = %d; the caller cannot gate on a figure that does not arrive", u.CacheRead) } - if !strings.Contains(string(srv.body), "THE-ASK") { + if !strings.Contains(string(srv.forwarded()), "THE-ASK") { t.Error("the ask did not reach the wire") } - if !strings.Contains(string(srv.body), "find the flaky test") { + if !strings.Contains(string(srv.forwarded()), "find the flaky test") { t.Error("the stashed transcript did not reach the wire, so there was no prefix to read") } // A DIFFERENT session must not read this one's prefix: it is another cache namespace, and diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index 956e5ccf..fe0ee5f3 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -7,6 +7,8 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -18,6 +20,69 @@ import ( "github.com/tidwall/gjson" ) +// upstreamRound is one request a fixture upstream served. +type upstreamRound struct { + method, path string + header http.Header + body []byte +} + +// upstreamCapture records what a fixture upstream saw, under a mutex. +// +// A fixture handler runs on the test server's own goroutine. The HTTP round trip that follows +// LOOKS like it orders the handler's writes before the test goroutine's reads, but it is not a +// happens-before edge: a captured variable written in the handler and read by the test after the +// round trip is an unsynchronised access, and `make cover` runs the suite with -race. Every field +// here therefore goes through u.mu, the shape tenancy_test.go's hostedFixture already uses. +// +// Counter-only fixtures do not need this: an atomic.Int64 carries its own edge and is a smaller +// change at the call site. +type upstreamCapture struct { + mu sync.Mutex + rounds []upstreamRound +} + +// record snapshots the request (draining its body) and returns the 1-based round number, so a +// handler that must answer differently per round reads its own count from the return value +// rather than from a captured counter. +func (u *upstreamCapture) record(r *http.Request) int { + b, _ := io.ReadAll(r.Body) + u.mu.Lock() + defer u.mu.Unlock() + u.rounds = append(u.rounds, upstreamRound{r.Method, r.URL.Path, r.Header.Clone(), b}) + return len(u.rounds) +} + +// hits is the number of rounds served. +func (u *upstreamCapture) hits() int { + u.mu.Lock() + defer u.mu.Unlock() + return len(u.rounds) +} + +// round returns the n-th round, 1-based; the zero value if there was no such round. +func (u *upstreamCapture) round(n int) upstreamRound { + u.mu.Lock() + defer u.mu.Unlock() + if n < 1 || n > len(u.rounds) { + return upstreamRound{} + } + return u.rounds[n-1] +} + +// body is the n-th round's body, 1-based; nil if there was no such round. +func (u *upstreamCapture) body(n int) []byte { return u.round(n).body } + +// last is the most recent round; the zero value if nothing was served. +func (u *upstreamCapture) last() upstreamRound { + u.mu.Lock() + defer u.mu.Unlock() + if len(u.rounds) == 0 { + return upstreamRound{} + } + return u.rounds[len(u.rounds)-1] +} + // buildHandler wires a real config->pipeline->proxy against a mock upstream that // records the body it receives. func buildHandler(t *testing.T, yaml string, upstream string) (*proxy.Handler, store.Store) { @@ -72,9 +137,9 @@ func expandableBody(hash string) []byte { } func TestProxyReducesThenForwards(t *testing.T) { - var got []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"ok":true}`)) })) @@ -98,6 +163,7 @@ func TestProxyReducesThenForwards(t *testing.T) { // Upstream must have received a SMALLER messages array (dedup collapsed the dup), // while non-message fields (model, temperature) survive verbatim (I1). + got := up.last().body if len(got) == 0 { t.Fatal("upstream received nothing") } @@ -117,9 +183,9 @@ func TestProxyReducesThenForwards(t *testing.T) { // gateway route with a Claude-Code-shaped body (tool outputs as tool_result // blocks in user messages) and asserts the offloader fires end-to-end. func TestAnthropicRouteReducesToolResult(t *testing.T) { - var got []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got, _ = io.ReadAll(r.Body) + up.record(r) w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"type":"message","content":[{"type":"text","text":"ok"}]}`)) })) @@ -148,6 +214,7 @@ func TestAnthropicRouteReducesToolResult(t *testing.T) { } resp.Body.Close() + got := up.last().body if len(got) == 0 { t.Fatal("upstream received nothing") } @@ -167,14 +234,9 @@ func TestAnthropicRouteReducesToolResult(t *testing.T) { // reduced like any chat and forwarded to the same path, while a control-plane // call (GET /admin/v1/profile) passes through to the upstream verbatim. func TestBobGatewayReducesModelAndPassesControlPlane(t *testing.T) { - type hit struct { - method, path string - body []byte - } - var hits []hit + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - b, _ := io.ReadAll(r.Body) - hits = append(hits, hit{r.Method, r.URL.Path, b}) + up.record(r) w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"ok":true}`)) })) @@ -213,10 +275,10 @@ func TestBobGatewayReducesModelAndPassesControlPlane(t *testing.T) { } cp.Body.Close() - if len(hits) != 2 { - t.Fatalf("want 2 upstream hits, got %d: %+v", len(hits), hits) + if n := up.hits(); n != 2 { + t.Fatalf("want 2 upstream hits, got %d", n) } - model, control := hits[0], hits[1] + model, control := up.round(1), up.round(2) if model.path != "/inference/v1/chat/completions" { t.Fatalf("model call forwarded to wrong path: %q", model.path) } @@ -235,9 +297,9 @@ func TestBobGatewayReducesModelAndPassesControlPlane(t *testing.T) { } func TestBypassHeaderForwardsUnchanged(t *testing.T) { - var got []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got, _ = io.ReadAll(r.Body) + up.record(r) w.Write([]byte(`{}`)) })) defer upstream.Close() @@ -257,16 +319,15 @@ func TestBypassHeaderForwardsUnchanged(t *testing.T) { t.Fatal(err) } resp.Body.Close() - if gjson.GetBytes(got, "messages.1.content").String() != dump { + if gjson.GetBytes(up.last().body, "messages.1.content").String() != dump { t.Fatal("bypass should forward messages unchanged") } } func TestGatewayInjectsRealKey(t *testing.T) { - var gotAuth, gotXAPI string + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotAuth = r.Header.Get("Authorization") - gotXAPI = r.Header.Get("x-api-key") + up.record(r) w.Write([]byte(`{}`)) })) defer upstream.Close() @@ -287,27 +348,25 @@ func TestGatewayInjectsRealKey(t *testing.T) { t.Fatal(err) } resp.Body.Close() - if gotAuth != "Bearer real-openai-key" { + hdr := up.last().header + if gotAuth := hdr.Get("Authorization"); gotAuth != "Bearer real-openai-key" { t.Fatalf("gateway should inject the real key, upstream saw %q", gotAuth) } - _ = gotXAPI + _ = hdr.Get("x-api-key") } func TestExpandToolLoop(t *testing.T) { - var calls int - var secondBody []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - b, _ := io.ReadAll(r.Body) - calls++ + round := up.record(r) w.Header().Set("Content-Type", "application/json") - if calls == 1 { + if round == 1 { // model asks to expand the offloaded content w.Write([]byte(`{"choices":[{"message":{"role":"assistant","tool_calls":[` + `{"id":"call_1","type":"function","function":{"name":"context_guru_expand","arguments":"{\"id\":\"HASH\"}"}}` + `]},"finish_reason":"tool_calls"}]}`)) return } - secondBody = b w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"done"}}]}`)) })) defer upstream.Close() @@ -325,12 +384,13 @@ func TestExpandToolLoop(t *testing.T) { final, _ := io.ReadAll(resp.Body) resp.Body.Close() - if calls != 2 { + if calls := up.hits(); calls != 2 { t.Fatalf("expected 2 upstream calls (initial + continuation), got %d", calls) } if !strings.Contains(string(final), "done") { t.Fatalf("proxy should return the final answer, got %s", final) } + secondBody := up.body(2) if !strings.Contains(string(secondBody), "THE ORIGINAL CONTENT") { t.Fatalf("continuation must carry the resolved original, got %s", secondBody) } @@ -344,12 +404,10 @@ func TestExpandToolLoop(t *testing.T) { // tool_use is context_guru_expand; the request carries a <> marker so the // proxy buffers+aggregates the SSE, resolves the original, and re-invokes upstream. func TestExpandSSELoop(t *testing.T) { - var calls int - var secondBody []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - b, _ := io.ReadAll(r.Body) - calls++ - if calls == 1 { + round := up.record(r) + if round == 1 { w.Header().Set("Content-Type", "text/event-stream") // A minimal Anthropic tool_use SSE: start, block start (tool_use), the input // json as one delta, stops. @@ -361,7 +419,6 @@ func TestExpandSSELoop(t *testing.T) { "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")) return } - secondBody = b w.Header().Set("Content-Type", "text/event-stream") w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"role\":\"assistant\"}}\n\n" + "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n" + @@ -397,9 +454,10 @@ func TestExpandSSELoop(t *testing.T) { final, _ := io.ReadAll(resp.Body) resp.Body.Close() - if calls != 2 { + if calls := up.hits(); calls != 2 { t.Fatalf("expected 2 upstream calls (SSE expand + continuation), got %d", calls) } + secondBody := up.body(2) if !strings.Contains(string(secondBody), "THE ORIGINAL CONTENT") { t.Fatalf("continuation must carry the resolved original, got %s", secondBody) } @@ -424,20 +482,17 @@ func TestExpandSSELoop(t *testing.T) { // must still carry a tool_result for BOTH call ids (the missing one gets a // placeholder) or the provider rejects the request. func TestExpandPartialResolutionWellFormed(t *testing.T) { - var calls int - var secondBody []byte + var up upstreamCapture upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - b, _ := io.ReadAll(r.Body) - calls++ + round := up.record(r) w.Header().Set("Content-Type", "application/json") - if calls == 1 { + if round == 1 { w.Write([]byte(`{"choices":[{"message":{"role":"assistant","tool_calls":[` + `{"id":"call_1","type":"function","function":{"name":"context_guru_expand","arguments":"{\"id\":\"GOOD\"}"}},` + `{"id":"call_2","type":"function","function":{"name":"context_guru_expand","arguments":"{\"id\":\"GONE\"}"}}` + `]},"finish_reason":"tool_calls"}]}`)) return } - secondBody = b w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`)) })) defer upstream.Close() @@ -454,9 +509,10 @@ func TestExpandPartialResolutionWellFormed(t *testing.T) { } resp.Body.Close() - if calls != 2 { + if calls := up.hits(); calls != 2 { t.Fatalf("expected a continuation round, got %d upstream calls", calls) } + secondBody := up.body(2) // One tool message per EXPAND tool_call_id (both call_1 and call_2), or the provider // errors. Counted by call id: the request already carried an unrelated tool turn (the // one holding the marker), which is not a result for this round. @@ -725,9 +781,9 @@ func TestMarkerBearingSSEStreamsWhenItOpensWithText(t *testing.T) { // answer only half a batch (the client owns Bash), so it must replay the stream // unchanged rather than continue — and the client's stream must stay well-formed. func TestExpandSSEWithOtherToolReplaysVerbatim(t *testing.T) { - var calls int + var calls atomic.Int64 // an atomic carries the happens-before edge the round trip does not upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - calls++ + calls.Add(1) w.Header().Set("Content-Type", "text/event-stream") w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"role\":\"assistant\"}}\n\n" + "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"context_guru_expand\"}}\n\n" + @@ -754,7 +810,7 @@ func TestExpandSSEWithOtherToolReplaysVerbatim(t *testing.T) { out, _ := io.ReadAll(resp.Body) resp.Body.Close() - if calls != 1 { + if calls := calls.Load(); calls != 1 { t.Fatalf("batched expand+Bash must NOT trigger a continuation, got %d upstream calls", calls) } // Verbatim replay: both blocks intact, indices unrenumbered, no injected content. @@ -770,10 +826,10 @@ func TestExpandSSEWithOtherToolReplaysVerbatim(t *testing.T) { // upstream that answers every request with another expand call must be cut off, and // the client must still get a well-formed stream (the model's own last call). func TestExpandSSEMultiRoundCapped(t *testing.T) { - var calls int + var calls atomic.Int64 // an atomic carries the happens-before edge the round trip does not upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.Copy(io.Discard, r.Body) - calls++ + calls.Add(1) w.Header().Set("Content-Type", "text/event-stream") w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"role\":\"assistant\"}}\n\n" + "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"context_guru_expand\"}}\n\n" + @@ -798,7 +854,7 @@ func TestExpandSSEMultiRoundCapped(t *testing.T) { resp.Body.Close() // maxExpandRounds continuations, then one final pass-through: 4 upstream calls. - if calls != 4 { + if calls := calls.Load(); calls != 4 { t.Fatalf("round cap not honored: %d upstream calls (want 4 = 3 rounds + terminal)", calls) } if !strings.Contains(string(out), "message_stop") { @@ -832,9 +888,9 @@ func TestExpandSSEMultiRoundCapped(t *testing.T) { // provider IS anthropic — a different branch from the provider gate at sse.go:21. // The client must still receive the original bytes unchanged. func TestExpandSSEAggregateFailureReplaysRaw(t *testing.T) { - var calls int + var calls atomic.Int64 // an atomic carries the happens-before edge the round trip does not upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - calls++ + calls.Add(1) w.Header().Set("Content-Type", "text/event-stream") // partial_json is TRUNCATED — it cannot reconstruct to valid JSON. w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"role\":\"assistant\"}}\n\n" + @@ -857,7 +913,7 @@ func TestExpandSSEAggregateFailureReplaysRaw(t *testing.T) { out, _ := io.ReadAll(resp.Body) resp.Body.Close() - if calls != 1 { + if calls := calls.Load(); calls != 1 { t.Fatalf("an unreconstructable stream must not drive a continuation: %d calls", calls) } if !strings.Contains(string(out), `\"id\":\"HA`) || !strings.Contains(string(out), "message_stop") { @@ -873,9 +929,9 @@ func TestExpandSSEAggregateFailureReplaysRaw(t *testing.T) { // marker-bearing OpenAI SSE response is replayed raw and restoration does not fire. // Correctness is preserved (fail-open); only the feature is absent. func TestExpandOpenAISSEFallsBackToRaw(t *testing.T) { - var calls int + var calls atomic.Int64 // an atomic carries the happens-before edge the round trip does not upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - calls++ + calls.Add(1) w.Header().Set("Content-Type", "text/event-stream") w.Write([]byte(`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"context_guru_expand","arguments":"{\"id\":\"HASH\"}"}}]}}]}` + "\n\n" + "data: [DONE]\n\n")) @@ -903,7 +959,7 @@ func TestExpandOpenAISSEFallsBackToRaw(t *testing.T) { out, _ := io.ReadAll(resp.Body) resp.Body.Close() - if calls != 1 { + if calls := calls.Load(); calls != 1 { t.Fatalf("OpenAI SSE cannot be aggregated, so no continuation is possible: %d calls", calls) } if !strings.Contains(string(out), "[DONE]") || strings.Contains(string(out), "THE ORIGINAL CONTENT") { diff --git a/proxy/tenancy_test.go b/proxy/tenancy_test.go index 59a45061..f9d473e1 100644 --- a/proxy/tenancy_test.go +++ b/proxy/tenancy_test.go @@ -488,9 +488,11 @@ func TestUnboundBobKeyRefusalNamesTheMissingCredential(t *testing.T) { // Single-tenant mode must be untouched: no token needed, static upstream used. func TestSingleTenantUnchanged(t *testing.T) { - var got *http.Request + // The handler runs on up's own goroutine; the round trip through h.Mux() is not a + // happens-before edge, so the clone comes back over a buffered channel. + seen := make(chan *http.Request, 1) up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - got = r.Clone(r.Context()) + seen <- r.Clone(r.Context()) w.Write([]byte(`{"ok":true}`)) })) defer up.Close() @@ -508,6 +510,11 @@ func TestSingleTenantUnchanged(t *testing.T) { t.Fatalf("single-tenant request = %d %s", w.Code, w.Body) } // With no key configured the client's own auth passes through, as documented. + var got *http.Request + select { + case got = <-seen: + default: + } if got == nil || got.Header.Get("Authorization") != "Bearer client-own-key" { t.Errorf("single-tenant pass-through changed: %v", got.Header.Get("Authorization")) } From 90b602bff4c9e9beb10377b5e09c51de7579025c Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Thu, 3 Sep 2026 14:12:14 +0300 Subject: [PATCH 2/2] test: address review on the fixture-capture synchronisation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six findings from the review of #191, in the two files the reviewer named. **expandsplice_test.go's remaining plain counter (review finding 1).** The body claimed `calls` at :136 was "read only inside the handler ... no cross-goroutine read". That does not hold: TestTheStreamedPrefixReachesTheClientBeforeTheExpandCall drives two upstream rounds and each is served from a goroutine of the httptest.Server's own, so the `calls++` of round 1 and the `calls++` / `if calls > 1` of round 2 touch one variable from two goroutines — sequential, but not ordered by anything the memory model gives you. Whether keep-alive reuse puts both rounds on the same conn goroutine is net/http's business, not an invariant a test may lean on. Converted to atomic.Int64, the shape already used five times in the same file, and the comment now states the actual reason rather than the false one. **The comment this PR made stale (finding 2).** :279 read "atomic, unlike the counters in the tests above" — the contrast died when those counters became atomics, and the last plain int among them was the site above. Reworded to stand on its own: the second round's connection is closed without a response, so not even a completed round trip sits between the write and the read. **captureUpstream duplicated upstreamCapture (inline, proxy_test.go:40).** modes_test.go's captureUpstream is a mutex-guarded body recorder in the same package, and upstreamCapture is a strict superset of it, so the package offered two overlapping ways to record an upstream. captureUpstream is now implemented over upstreamCapture via a new bodies() accessor; its narrow signature stays because its fifteen call sites only want the bodies in order. `sync` drops out of modes_test.go's imports. The reasoning for the synchronisation now lives in exactly one place. **The dead x-api-key statement (inline, proxy_test.go:355).** `_ = hdr.Get("x-api-key")` carried over from an equally dead `_ = gotXAPI` and asserted nothing. Written as the assertion it was presumably reaching for: an OpenAI upstream must not receive Anthropic's header, least of all a copy of the key. **The lost failure diagnostic (inline, proxy_test.go:279).** The pre-PR message dumped the hits; the PR reduced it to the count, which is the least informative part. Restored via a served() accessor — plus a String() on upstreamRound, because %+v on a []byte field renders the body as a list of decimal byte values and defeats the point of dumping it. **The nil dereference on the path the select exists to preserve (inline, tenancy_test.go:514).** TestSingleTenantUnchanged's non-blocking receive is there so "nothing arrived" stays a reportable failure instead of a deadlock — but the t.Errorf argument was evaluated unconditionally, so got.Header on a nil *http.Request panicked and took the whole test binary down. Pre-existing and unchanged by #191, but #191 is what turned that path from can't-happen into designed-for, so it is fixed here: t.Fatal on nil before the header is read. Verification (eval box, go1.26.4, CGO_ENABLED=1): - `gofmt -l proxy` clean, `go vet ./proxy/` clean, `go build ./...` clean. - `go test -race ./proxy/ ./internal/cheapmodel/` passing. - The two new assertions were revert-verified rather than assumed: - x-api-key: mutating the OpenAI route's setKey to also set the header (`hd.Set("x-api-key", h.opts.OpenAIKey)`) fails as `proxy_test.go:377: gateway sent x-api-key to an OpenAI upstream: "real-openai-key"`. Restored, passes. - the nil guard: dropping the handler's `seen <- r.Clone(...)` to simulate a request that never arrives fails cleanly as `tenancy_test.go:517: single-tenant request never reached the upstream`, where the pre-fix assertion panics at tenancy_test.go:519 inside t.Errorf's argument. Restored, passes. - The atomic conversion is not revert-verifiable, for the reason #191 already documents at length: -race does not fire on this shape before or after, so it is a finding by inspection. Unrelated pre-existing failure found while re-running the suite at `-count=2` and reported as #192 rather than patched here: TestExtractEconomicsAreExported and TestExpandUnresolvedSeriesRender assert on process-global metrics counters they increment themselves, so the proxy package cannot pass at -count>1. Reproduces identically on unmodified origin/main, so it belongs on its own branch. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: DAVID AMID --- proxy/expandsplice_test.go | 15 +++++++++------ proxy/modes_test.go | 19 ++++++------------- proxy/proxy_test.go | 34 ++++++++++++++++++++++++++++++++-- proxy/tenancy_test.go | 5 +++-- 4 files changed, 50 insertions(+), 23 deletions(-) diff --git a/proxy/expandsplice_test.go b/proxy/expandsplice_test.go index 8ecf6052..b4fc1454 100644 --- a/proxy/expandsplice_test.go +++ b/proxy/expandsplice_test.go @@ -133,13 +133,16 @@ func TestExpandCalledAfterALeadingBlockIsNeverGivenToTheClient(t *testing.T) { // cannot pass this, which is the whole difference between splicing and re-buffering. func TestTheStreamedPrefixReachesTheClientBeforeTheExpandCall(t *testing.T) { release := make(chan struct{}) - var calls int + // Two rounds, each served from a goroutine of upstream's own: the counter is read on a + // different goroutine from the one that last wrote it, never mind that the proxy issues the + // rounds one after the other. Which conn goroutine serves round 2 is net/http's business, + // not something the test may assume, so the counter carries its own edge. + var calls atomic.Int64 head, tail := leadThenExpand("text") upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.Copy(io.Discard, r.Body) - calls++ w.Header().Set("Content-Type", "text/event-stream") - if calls > 1 { + if calls.Add(1) > 1 { w.Write([]byte(round2Answer)) return } @@ -276,9 +279,9 @@ func TestTheClientsNoSuchToolErrorNeverReachesTheModel(t *testing.T) { // while the response was buffered and nothing had been written, and is garbage appended to // the model's turn now that the prefix has already streamed. func TestAFailedContinuationRoundStillEndsTheClientsTurn(t *testing.T) { - // atomic, unlike the counters in the tests above: the second round's connection is closed - // without a response, so there is no happens-before edge between the handler's write and - // the assertion's read. + // The second round's connection is closed without a response, so not even a completed round + // trip stands between the handler's write and the assertion's read; the atomic carries the + // edge itself. var calls atomic.Int64 head, tail := leadThenExpand("text") upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/proxy/modes_test.go b/proxy/modes_test.go index c67bef05..a6670ea9 100644 --- a/proxy/modes_test.go +++ b/proxy/modes_test.go @@ -9,7 +9,6 @@ import ( "runtime" "strconv" "strings" - "sync" "sync/atomic" "testing" "time" @@ -50,25 +49,19 @@ func newModeHandler(t *testing.T, yaml, upstream string, mode components.Mode, c return h, agg } -// captureUpstream records every body the upstream receives. +// captureUpstream records every body the upstream receives. It is the bodies-only view over +// upstreamCapture (proxy_test.go), which is where the synchronisation and the reasoning for it +// live; the narrow signature stays because its call sites only ever want the bodies in order. func captureUpstream(t *testing.T) (*httptest.Server, func() [][]byte) { t.Helper() - var mu sync.Mutex - var got [][]byte + var up upstreamCapture srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - b, _ := io.ReadAll(r.Body) - mu.Lock() - got = append(got, b) - mu.Unlock() + up.record(r) w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"ok":true}`)) })) t.Cleanup(srv.Close) - return srv, func() [][]byte { - mu.Lock() - defer mu.Unlock() - return append([][]byte(nil), got...) - } + return srv, up.bodies } const modePipeline = "pipeline: [dedup, cacheinject]\n" diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index fe0ee5f3..b97dbd93 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -3,6 +3,7 @@ package proxy_test import ( "bytes" "encoding/json" + "fmt" "io" "net/http" "net/http/httptest" @@ -27,6 +28,12 @@ type upstreamRound struct { body []byte } +// String renders the round for a failure message with the body as text. Plain %+v on a []byte +// field prints a list of decimal byte values, which defeats the point of dumping it at all. +func (r upstreamRound) String() string { + return fmt.Sprintf("%s %s header=%v body=%s", r.method, r.path, r.header, r.body) +} + // upstreamCapture records what a fixture upstream saw, under a mutex. // // A fixture handler runs on the test server's own goroutine. The HTTP round trip that follows @@ -73,6 +80,25 @@ func (u *upstreamCapture) round(n int) upstreamRound { // body is the n-th round's body, 1-based; nil if there was no such round. func (u *upstreamCapture) body(n int) []byte { return u.round(n).body } +// served is a copy of every round, in order, for failure messages that should say what did +// arrive rather than only how many things did. +func (u *upstreamCapture) served() []upstreamRound { + u.mu.Lock() + defer u.mu.Unlock() + return append([]upstreamRound(nil), u.rounds...) +} + +// bodies is the served bodies in order — the view captureUpstream hands its callers. +func (u *upstreamCapture) bodies() [][]byte { + u.mu.Lock() + defer u.mu.Unlock() + out := make([][]byte, len(u.rounds)) + for i, r := range u.rounds { + out[i] = r.body + } + return out +} + // last is the most recent round; the zero value if nothing was served. func (u *upstreamCapture) last() upstreamRound { u.mu.Lock() @@ -276,7 +302,7 @@ func TestBobGatewayReducesModelAndPassesControlPlane(t *testing.T) { cp.Body.Close() if n := up.hits(); n != 2 { - t.Fatalf("want 2 upstream hits, got %d", n) + t.Fatalf("want 2 upstream hits, got %d: %+v", n, up.served()) } model, control := up.round(1), up.round(2) if model.path != "/inference/v1/chat/completions" { @@ -352,7 +378,11 @@ func TestGatewayInjectsRealKey(t *testing.T) { if gotAuth := hdr.Get("Authorization"); gotAuth != "Bearer real-openai-key" { t.Fatalf("gateway should inject the real key, upstream saw %q", gotAuth) } - _ = hdr.Get("x-api-key") + // And only that one slot: an OpenAI upstream has no business receiving Anthropic's header, + // least of all a copy of the key. + if gotXAPI := hdr.Get("x-api-key"); gotXAPI != "" { + t.Fatalf("gateway sent x-api-key to an OpenAI upstream: %q", gotXAPI) + } } func TestExpandToolLoop(t *testing.T) { diff --git a/proxy/tenancy_test.go b/proxy/tenancy_test.go index f9d473e1..a9d7ac81 100644 --- a/proxy/tenancy_test.go +++ b/proxy/tenancy_test.go @@ -514,9 +514,10 @@ func TestSingleTenantUnchanged(t *testing.T) { select { case got = <-seen: default: + t.Fatal("single-tenant request never reached the upstream") } - if got == nil || got.Header.Get("Authorization") != "Bearer client-own-key" { - t.Errorf("single-tenant pass-through changed: %v", got.Header.Get("Authorization")) + if auth := got.Header.Get("Authorization"); auth != "Bearer client-own-key" { + t.Errorf("single-tenant pass-through changed: %v", auth) } }