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