From 5d82a6ce656bcf6b7a7a8a7dbcbb47921750ac95 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 3 Sep 2026 08:56:48 +0100 Subject: [PATCH] fix(target-health): record a hung federated server as a failure The previous round suppressed every context-caused federation error. A server that accepted the connection and then never answered therefore read "unknown" forever whenever federation carried the only traffic, which is the outage shape this feature exists to surface. Split the two causes instead. A server still waiting for a fan-out slot when the budget expires is never dialed, so the semaphore acquire now selects on the fan-out context and records nothing for it. A server this gateway did dial records a failure on a deadline, because it had its chance to answer. Only a caller hanging up stays unattributable. Document the split rule, correct a stale comment about a second record site, and state that "unknown" reads two ways rather than three. Claude-Session: https://claude.ai/code/session_01EebV5jyWvYocByNQ3oss1w --- README.md | 23 +++++---- mcp_federation.go | 69 +++++++++++++++++-------- mcp_federation_test.go | 113 ++++++++++++++++++++++++++++++++++------- 3 files changed, 156 insertions(+), 49 deletions(-) diff --git a/README.md b/README.md index 49b826b..a6324e4 100644 --- a/README.md +++ b/README.md @@ -1898,11 +1898,14 @@ plugin's config. only when the outbound call itself fails (`err != nil` from the backend call) — a JSON-RPC-level `error` in the backend's own response still counts as reachable, since the server answered. A - context-caused failure (the caller canceling, or the shared fan-out - deadline expiring) records nothing at all, the same client-cancel - rule the per-server proxy above applies; a genuinely hung server is - still caught by the active probe sweep or its own per-server proxy - health. + context-caused failure is split two ways: the caller hanging up + (`context.Canceled`) records nothing, the same client-cancel rule the + per-server proxy above applies, but a server this gateway actually + dialed that never answers before the shared fan-out budget expires + (`context.DeadlineExceeded`) records a real failure — it had a chance + to answer and did not. A server still waiting for a free fan-out slot + when that same budget expires — never dialed at all — records + nothing, since this gateway never gave it the chance. - **Active probes are opt-in** (`targetHealth.enabled: true`) and lazy: no timer, no background goroutine running on its own. A probe sweep is triggered by `/metrics`, `GET /admin/api/targets`, `GET @@ -1929,8 +1932,7 @@ plugin's config. whether or not the target has ever succeeded). A target with one or two recent failures after a success still reads `"healthy"` until the threshold is actually crossed. - - **`"unknown"` reads three ways**, all sharing one JSON `state` - value: + - **`"unknown"` reads two ways**, both sharing one JSON `state` value: 1. never observed at all — no probe or passive traffic has reached this target yet. `GET /admin/api/targets` omits every health field but `state`/`consecutiveFailures` for this case. @@ -1941,9 +1943,10 @@ plugin's config. has never seen answer is not `"healthy"` either. Every other health field is still present here, since there is a real observation to show. - 3. either way, `llmgateway_target_healthy` emits no sample while - `state` is `"unknown"` — a gap in the series is honest, a - fabricated `1` is not. + + Either way, `llmgateway_target_healthy` emits no sample while `state` + is `"unknown"` — a gap in the series is honest, a fabricated `1` is + not. ## Security notes diff --git a/mcp_federation.go b/mcp_federation.go index bdd6bd2..17fe2f6 100644 --- a/mcp_federation.go +++ b/mcp_federation.go @@ -970,28 +970,53 @@ func (g *Gateway) mcpFederatedToolsList(w http.ResponseWriter, format mcpRespons } }() - sem <- struct{}{} + // G4, feat/target-health review round 2: acquiring the + // semaphore via select, not a bare blocking send, lets a + // server still parked here when fanoutCtx expires return + // WITHOUT ever calling mcpBackendCall — it was never dialed, + // so it has nothing to say about that server's own health + // (below). Without this, a bare `sem <- struct{}{}` still + // unblocks once fanoutCtx dies (an earlier holder's own call + // fails fast against the already-expired context and frees + // its slot), but by then mcpBackendCall would be called with + // a context that is already done — indistinguishable from a + // server that WAS dialed and genuinely timed out, which is + // exactly the over-suppression this fix removes. + select { + case sem <- struct{}{}: + case <-fanoutCtx.Done(): + mu.Lock() + failed = append(failed, name) + mu.Unlock() + return + } defer func() { <-sem }() targetURL := g.cfg.MCPServers[name].URL // feat/target-health: passive recording (target_health.go) — // err != nil is a failure; a JSON-RPC-level resp.Error is NOT - // (the server answered, it just reported its own error), so - // it is recorded separately, below, once the shape is known. + // (the server answered, it just reported its own error). The + // record call right below already covers both: it reads ok + // from err == nil alone, so a non-nil resp.Error alongside a + // nil err still records success here — the failed-list + // append further down (used only for the merged tools/list + // result and the loud-failure short-circuit) is a separate + // concern and never touches targetHealth itself. probeStart := time.Now() resp, err := g.mcpBackendCall(fanoutCtx, targetURL, "tools/list", struct{}{}, mcpBackendResponseMaxBytes) - // F2: fanoutCtx is derived from r.Context() (above), so a - // client hang-up (context.Canceled) surfaces here as EVERY - // still-in-flight server's own error, and the shared - // toolsListBackendTimeout expiry (context.DeadlineExceeded) - // surfaces identically for a server that was genuinely slow - // AND one still parked on sem, never dialed at all — neither - // says anything about that server's own health, so neither is - // recorded, mirroring recordTargetProxyHealth's identical - // client-cancel rule (target_health.go). A server that is - // really hung is instead caught by the active probe sweep or - // its own per-request proxy health, not by this fan-out. - if err == nil || (!errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)) { + // G4: this goroutine WON the select above, so it was actually + // dialed — only a client hang-up (context.Canceled, + // propagating from r.Context() into fanoutCtx) says nothing + // about this server's own health, mirroring + // recordTargetProxyHealth's identical client-cancel rule + // (target_health.go). context.DeadlineExceeded here means the + // shared toolsListBackendTimeout fired while THIS dialed + // server was still in flight — it did not answer within the + // fan-out's own budget, which is a real failure worth + // recording, not a client artifact (F2 over-suppressed this + // case; a genuinely hung server used to read "unknown" + // forever under federation-only traffic). + if err == nil || !errors.Is(err, context.Canceled) { g.targetHealth.record(targetKindMCP, name, targetURL, err == nil, err, time.Since(probeStart), targetHealthSourceTraffic) } @@ -1116,11 +1141,15 @@ func (g *Gateway) mcpFederatedToolsCall(w http.ResponseWriter, format mcpRespons // not. probeStart := time.Now() resp, err := g.mcpBackendCall(ctx, targetURL, "tools/call", mcpToolCallParams{Name: toolName, Arguments: params.Arguments}, mcpBackendCallResponseMaxBytes) - // F2: same context-caused-error guard as mcpFederatedToolsList's own - // fan-out, above — ctx here derives from r.Context() too, so a client - // cancel or the shared toolsCallBackendTimeout expiry must record - // nothing rather than a false failure. - if err == nil || (!errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)) { + // G4: same dialed-server rule as mcpFederatedToolsList's own fan-out, + // above — ctx here derives from r.Context() too, so a client cancel + // (context.Canceled) must record nothing. Unlike the fan-out, + // tools/call has no semaphore to wait behind: the call above is + // always actually dialed, so context.DeadlineExceeded here means + // THIS server did not answer within toolsCallBackendTimeout — a real + // failure worth recording, not a client artifact (F2 over-suppressed + // this case). + if err == nil || !errors.Is(err, context.Canceled) { g.targetHealth.record(targetKindMCP, serverName, targetURL, err == nil, err, time.Since(probeStart), targetHealthSourceTraffic) } g.limiter.countTargetRequest(targetKindMCP, serverName) diff --git a/mcp_federation_test.go b/mcp_federation_test.go index a386eb6..9fc1f41 100644 --- a/mcp_federation_test.go +++ b/mcp_federation_test.go @@ -858,10 +858,12 @@ func TestHandleMCPFederated_ToolsCall_SlowBackend_BoundedByRequestContext(t *tes t.Cleanup(func() { close(block) }) cfg := newFederationTestConfig(srv.URL, "http://beta.invalid", true) + cfg.TargetHealth = TargetHealthConfig{FailureThreshold: 1} h, err := New(context.Background(), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } + gw := h.(*Gateway) callParams, _ := json.Marshal(mcpToolCallParams{Name: "alpha_lookup"}) req := newFederatedRequest(t, "sk-alice", jsonrpcRequest{JSONRPC: jsonrpcVersion, Method: "tools/call", ID: json.RawMessage("1"), Params: callParams}) @@ -884,6 +886,12 @@ func TestHandleMCPFederated_ToolsCall_SlowBackend_BoundedByRequestContext(t *tes if got.Error == nil || got.Error.Code != jsonrpcInternalError { t.Errorf("error = %+v, want internal error (the resolved server timed out)", got.Error) } + // G4, feat/target-health review round 2: tools/call always actually + // dials its one resolved server (no semaphore to wait behind), so + // this shared-budget expiry must record a failure, not nothing. + if state := gw.targetHealth.stateOf(targetKindMCP, "alpha"); state != targetHealthUnhealthy { + t.Errorf("alpha state = %q, want unhealthy (dialed and never answered within the request's own budget)", state) + } } // --- MF2: non-2xx status and empty-envelope guards --- @@ -2057,28 +2065,72 @@ func TestHandleMCPFederated_ToolsList_ClientCanceled_RecordsNothing(t *testing.T } } -// TestHandleMCPFederated_ToolsList_SharedDeadlineExpiry_RecordsNothing -// proves F2's second case: when the fan-out's own shared context expires -// (toolsListBackendTimeout, or here the incoming request's own shorter -// deadline — see TestHandleMCPFederated_ToolsList_SlowBackend_BoundedByRequestContext's -// own doc comment for why that stands in for the real 20s budget), a -// server still parked behind the fan-out semaphore — never actually -// dialed — must not be recorded as a failure either. mcpBackendCall -// returns the identical context.DeadlineExceeded whether a server was -// slow or never contacted at all, so this case is handled by the same -// guard as the client-cancel case above, not by tracking who was dialed. -func TestHandleMCPFederated_ToolsList_SharedDeadlineExpiry_RecordsNothing(t *testing.T) { +// TestHandleMCPFederated_ToolsList_HungDialedServer_RecordsFailure proves +// G4 (feat/target-health review round 2): a server that WAS actually +// dialed and never answers before the fan-out's own shared budget +// expires (toolsListBackendTimeout, or here the incoming request's own +// shorter deadline — see +// TestHandleMCPFederated_ToolsList_SlowBackend_BoundedByRequestContext's +// own doc comment for why that stands in for the real 20s budget) must +// still record a failure, not nothing. F2's original guard +// blanket-suppressed context.DeadlineExceeded too, which meant a +// genuinely hung server read "unknown" forever under federation-only +// traffic whenever targetHealth.enabled defaults false — the +// over-suppression this fix removes. +func TestHandleMCPFederated_ToolsList_HungDialedServer_RecordsFailure(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-release + })) + t.Cleanup(srv.Close) + t.Cleanup(func() { close(release) }) + + cfg := newFederationTestConfig(srv.URL, "http://beta.invalid", true) + cfg.TargetHealth = TargetHealthConfig{FailureThreshold: 1} + h, err := New(context.Background(), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), cfg, "llmgw") + if err != nil { + t.Fatalf("New: %v", err) + } + gw := h.(*Gateway) + + req := newFederatedRequest(t, "sk-alice", jsonrpcRequest{JSONRPC: jsonrpcVersion, Method: "tools/list", ID: json.RawMessage("1")}) + shortCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + req = req.WithContext(shortCtx) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + if got := gw.targetHealth.stateOf(targetKindMCP, "alpha"); got != targetHealthUnhealthy { + t.Errorf("alpha state = %q, want unhealthy (dialed and never answered within the fan-out budget)", got) + } +} + +// TestHandleMCPFederated_ToolsList_NeverDialedSemaphoreStarved_RecordsNothing +// proves G4's other half: a server still parked behind the fan-out +// semaphore when the shared deadline fires — never actually dialed — +// must still record nothing, unlike a server that WAS dialed (the test +// above). One more server than the fan-out semaphore has slots for +// (mcpFederatedFanoutConcurrency) guarantees at least one is never +// dialed; draining exactly mcpFederatedFanoutConcurrency sends on +// reached first proves which servers WERE actually dialed before any +// assertion runs, rather than assuming which specific names win the +// race. snapshot's own "observed" field (target_health.go), not stateOf, +// is what actually distinguishes "recorded a failure, still below +// failureThreshold" from "never recorded at all" — both read state +// targetHealthUnknown (F4), so FailureThreshold: 1 below also makes a +// dialed server's single failure visible as targetHealthUnhealthy +// through stateOf too, for a belt-and-suspenders check. +func TestHandleMCPFederated_ToolsList_NeverDialedSemaphoreStarved_RecordsNothing(t *testing.T) { + reached := make(chan struct{}, mcpFederatedFanoutConcurrency) block := make(chan struct{}) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached <- struct{}{} <-block })) t.Cleanup(srv.Close) t.Cleanup(func() { close(block) }) - // One more server than the fan-out semaphore has slots for - // (mcpFederatedFanoutConcurrency) guarantees at least one is still - // parked on the semaphore, never dialed, when the shared deadline - // fires. names := make([]string, 0, mcpFederatedFanoutConcurrency+1) mcpServers := make(map[string]*TargetConfig, mcpFederatedFanoutConcurrency+1) for i := 0; i < mcpFederatedFanoutConcurrency+1; i++ { @@ -2092,6 +2144,7 @@ func TestHandleMCPFederated_ToolsList_SharedDeadlineExpiry_RecordsNothing(t *tes cfg.MCPServers = mcpServers cfg.Groups = map[string]*GroupConfig{"default": {}} cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice"}}} + cfg.TargetHealth = TargetHealthConfig{FailureThreshold: 1} h, err := New(context.Background(), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), cfg, "llmgw") if err != nil { @@ -2100,17 +2153,39 @@ func TestHandleMCPFederated_ToolsList_SharedDeadlineExpiry_RecordsNothing(t *tes gw := h.(*Gateway) req := newFederatedRequest(t, "sk-alice", jsonrpcRequest{JSONRPC: jsonrpcVersion, Method: "tools/list", ID: json.RawMessage("1")}) - shortCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + shortCtx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) defer cancel() req = req.WithContext(shortCtx) rec := httptest.NewRecorder() - h.ServeHTTP(rec, req) + done := make(chan struct{}) + go func() { + h.ServeHTTP(rec, req) + close(done) + }() + for i := 0; i < mcpFederatedFanoutConcurrency; i++ { + <-reached + } + <-done + + dialed, neverDialed := 0, 0 for _, name := range names { - if got := gw.targetHealth.stateOf(targetKindMCP, name); got != targetHealthUnknown { - t.Errorf("mcp/%s state = %q, want unknown (a shared fan-out deadline expiry must record nothing)", name, got) + snap := gw.targetHealth.snapshot(targetKindMCP, name) + if snap.observed { + dialed++ + if snap.state != targetHealthUnhealthy { + t.Errorf("mcp/%s state = %q, want unhealthy (dialed and never answered)", name, snap.state) + } + continue } + neverDialed++ + } + if dialed != mcpFederatedFanoutConcurrency { + t.Errorf("dialed = %d, want %d (exactly one server must never have been dialed)", dialed, mcpFederatedFanoutConcurrency) + } + if neverDialed != 1 { + t.Errorf("neverDialed = %d, want exactly 1", neverDialed) } }