fix(transport): cap plugin MCP startup at ~4s when endpoints are unreachable
Issue #119: an unreachable third-party MCP plugin (e.g. blocked/firewalled endpoint) blocks `dws --help` for ~10s on every CLI invocation, because plugin discovery happens eagerly during command-tree construction. Root cause was a stack of timeouts that multiplied under transport failure: * `transport.Client.Initialize` loops three supported protocol versions on every error — including dial timeouts and HTTP 5xx — even though those failure modes are protocol-version-independent. Three loops × the dial budget = 3× the worst-case startup cost. * Default `DialContext.Timeout` was 10s, so a single dial against an unroutable address (e.g. TEST-NET-1) burned the full plugin context. * `registerHTTPServer` granted plugins with `AuthHeaders` a 10s outer context (intended for slow third-party services), and `registerStdioServer` granted every stdio plugin 10s. Either single misbehaving plugin therefore stalled the entire CLI. * Registry-side `defaultDiscoveryTimeout` (10s) and `perServerDiscoveryTimeout` (5s) had the same shape on the LoadCatalog path. Changes: * `transport.Client.Initialize`: short-circuit when the underlying `*CallError.Stage` is anything other than `CallStageJSONRPC`. Protocol version negotiation is the only justification for retrying with another version; transport/HTTP failures fail identically and should surrender. * `transport.defaultTransport`: `DialContext.Timeout` 10s → 3s. * `app.registerHTTPServer`: AuthHeaders timeout 10s → 4s. * `app.registerStdioServer`: outer ctx timeout 10s → 4s. * `cli.defaultDiscoveryTimeout`: 10s → 4s. * `discovery.Service`: rename `perServerDiscoveryTimeout` 5s → 2s (`defaultPerServerDiscoveryTimeout`); add `Service.PerServerTimeout` override field for tests/callers needing a tighter or looser bound. Tests: * `TestInitializeShortCircuitsOnHTTPError` — asserts only ONE protocol version is attempted on HTTP 5xx. * `TestInitializeShortCircuitsOnDialFailure` — asserts Initialize returns in <2s on a refused-connection address. * Existing `TestInitializeNegotiatesProtocolVersion` continues to pass — JSON-RPC-stage errors still trigger version fallback. End-to-end measurement with a blackhole plugin (endpoint=192.0.2.1, AuthHeaders set): | Variant | median `dws --help` | | ------------- | ------------------- | | main | 10.2s | | PR #121 alone | 9.1s | | this branch | 3.72s (-63%) | The `loader.go` and `discovery/service.go` timeout reductions overlap with PR #121 by @utafrali; see PR description for attribution. Refs: #119 Supersedes: #121
This commit is contained in:
+10
-2
@@ -1193,9 +1193,14 @@ func loadPlugins(engine *pipeline.Engine, runner executor.Runner) []*cobra.Comma
|
|||||||
func registerHTTPServer(p *plugin.Plugin, srv market.ServerDescriptor, tc *transport.Client, runner executor.Runner) []*cobra.Command {
|
func registerHTTPServer(p *plugin.Plugin, srv market.ServerDescriptor, tc *transport.Client, runner executor.Runner) []*cobra.Command {
|
||||||
// Use a longer timeout for servers with custom auth headers (third-party
|
// Use a longer timeout for servers with custom auth headers (third-party
|
||||||
// services may have higher latency than local/DingTalk endpoints).
|
// services may have higher latency than local/DingTalk endpoints).
|
||||||
|
// Bounded tightly so an unreachable plugin cannot stall every CLI
|
||||||
|
// command: combined with DialContext.Timeout=3s plus Initialize's
|
||||||
|
// short-circuit on transport errors, 4s still accommodates a healthy
|
||||||
|
// third-party endpoint (typically <2s) while surrendering quickly on
|
||||||
|
// broken ones. See issue #119.
|
||||||
timeout := 2 * time.Second
|
timeout := 2 * time.Second
|
||||||
if len(srv.AuthHeaders) > 0 {
|
if len(srv.AuthHeaders) > 0 {
|
||||||
timeout = 10 * time.Second
|
timeout = 4 * time.Second
|
||||||
}
|
}
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -1342,7 +1347,10 @@ func registerPluginAuthFromHeaders(srv market.ServerDescriptor) {
|
|||||||
// via ListTools, builds CLI commands, and registers the StdioClient for
|
// via ListTools, builds CLI commands, and registers the StdioClient for
|
||||||
// runtime dispatch. Returns generated cobra commands.
|
// runtime dispatch. Returns generated cobra commands.
|
||||||
func registerStdioServer(p *plugin.Plugin, sc plugin.StdioServerClient, runner executor.Runner) []*cobra.Command {
|
func registerStdioServer(p *plugin.Plugin, sc plugin.StdioServerClient, runner executor.Runner) []*cobra.Command {
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
// Bounded startup window for stdio plugins. A healthy MCP subprocess
|
||||||
|
// initialises well under a second; capping at 4s keeps a misbehaving
|
||||||
|
// plugin from blocking every CLI command. See issue #119.
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
if _, err := sc.Client.Initialize(ctx); err != nil {
|
if _, err := sc.Client.Initialize(ctx); err != nil {
|
||||||
|
|||||||
@@ -107,7 +107,9 @@ const (
|
|||||||
DefaultMarketBaseURL = "https://mcp.dingtalk.com"
|
DefaultMarketBaseURL = "https://mcp.dingtalk.com"
|
||||||
|
|
||||||
// defaultDiscoveryTimeout bounds the time spent on live registry discovery.
|
// defaultDiscoveryTimeout bounds the time spent on live registry discovery.
|
||||||
defaultDiscoveryTimeout = 10 * time.Second
|
// Tightened to 4s so a slow/unreachable discovery endpoint cannot block
|
||||||
|
// every CLI command invocation. See issue #119.
|
||||||
|
defaultDiscoveryTimeout = 4 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
type CatalogLoader interface {
|
type CatalogLoader interface {
|
||||||
|
|||||||
@@ -59,6 +59,11 @@ type Service struct {
|
|||||||
Tenant string
|
Tenant string
|
||||||
AuthIdentity string
|
AuthIdentity string
|
||||||
Logger *slog.Logger
|
Logger *slog.Logger
|
||||||
|
// PerServerTimeout overrides the default per-server discovery timeout
|
||||||
|
// when greater than zero. Useful for tests and for callers that need a
|
||||||
|
// tighter or looser bound. When zero, defaultPerServerDiscoveryTimeout
|
||||||
|
// applies.
|
||||||
|
PerServerTimeout time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
type RuntimeServer struct {
|
type RuntimeServer struct {
|
||||||
@@ -170,7 +175,11 @@ func (s *Service) DiscoverServerRuntime(ctx context.Context, server market.Serve
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
const perServerDiscoveryTimeout = 5 * time.Second
|
// defaultPerServerDiscoveryTimeout bounds the time spent discovering tools on
|
||||||
|
// a single registry-listed server. Tightened to 2s so a slow/unreachable
|
||||||
|
// server cannot stall every CLI command — a healthy MCP endpoint negotiates
|
||||||
|
// well under a second. See issue #119.
|
||||||
|
const defaultPerServerDiscoveryTimeout = 2 * time.Second
|
||||||
|
|
||||||
func (s *Service) DiscoverAllRuntime(ctx context.Context, servers []market.ServerDescriptor) ([]RuntimeServer, []RuntimeFailure) {
|
func (s *Service) DiscoverAllRuntime(ctx context.Context, servers []market.ServerDescriptor) ([]RuntimeServer, []RuntimeFailure) {
|
||||||
type discoveryResult struct {
|
type discoveryResult struct {
|
||||||
@@ -178,6 +187,11 @@ func (s *Service) DiscoverAllRuntime(ctx context.Context, servers []market.Serve
|
|||||||
failure *RuntimeFailure
|
failure *RuntimeFailure
|
||||||
}
|
}
|
||||||
|
|
||||||
|
perServerTimeout := defaultPerServerDiscoveryTimeout
|
||||||
|
if s.PerServerTimeout > 0 {
|
||||||
|
perServerTimeout = s.PerServerTimeout
|
||||||
|
}
|
||||||
|
|
||||||
filtered := make([]market.ServerDescriptor, 0, len(servers))
|
filtered := make([]market.ServerDescriptor, 0, len(servers))
|
||||||
for _, srv := range servers {
|
for _, srv := range servers {
|
||||||
if !srv.CLI.Skip {
|
if !srv.CLI.Skip {
|
||||||
@@ -194,7 +208,7 @@ func (s *Service) DiscoverAllRuntime(ctx context.Context, servers []market.Serve
|
|||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func(server market.ServerDescriptor) {
|
go func(server market.ServerDescriptor) {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
serverCtx, cancel := context.WithTimeout(ctx, perServerDiscoveryTimeout)
|
serverCtx, cancel := context.WithTimeout(ctx, perServerTimeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
rs, err := s.DiscoverServerRuntime(serverCtx, server)
|
rs, err := s.DiscoverServerRuntime(serverCtx, server)
|
||||||
|
|||||||
@@ -212,7 +212,7 @@ func (r *ToolCallResult) UnmarshalJSON(data []byte) error {
|
|||||||
func defaultTransport() *http.Transport {
|
func defaultTransport() *http.Transport {
|
||||||
return &http.Transport{
|
return &http.Transport{
|
||||||
DialContext: (&net.Dialer{
|
DialContext: (&net.Dialer{
|
||||||
Timeout: 10 * time.Second,
|
Timeout: 3 * time.Second,
|
||||||
KeepAlive: 30 * time.Second,
|
KeepAlive: 30 * time.Second,
|
||||||
}).DialContext,
|
}).DialContext,
|
||||||
TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12},
|
TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12},
|
||||||
@@ -298,6 +298,7 @@ func SupportedProtocolVersions() []string {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *Client) Initialize(ctx context.Context, endpoint string) (InitializeResult, error) {
|
func (c *Client) Initialize(ctx context.Context, endpoint string) (InitializeResult, error) {
|
||||||
|
var lastErr error
|
||||||
for _, version := range SupportedProtocolVersions() {
|
for _, version := range SupportedProtocolVersions() {
|
||||||
params := map[string]any{
|
params := map[string]any{
|
||||||
"capabilities": map[string]any{},
|
"capabilities": map[string]any{},
|
||||||
@@ -309,18 +310,33 @@ func (c *Client) Initialize(ctx context.Context, endpoint string) (InitializeRes
|
|||||||
}
|
}
|
||||||
|
|
||||||
var payload InitializeResult
|
var payload InitializeResult
|
||||||
if err := c.callJSONRPC(ctx, endpoint, requestEnvelope{
|
err := c.callJSONRPC(ctx, endpoint, requestEnvelope{
|
||||||
JSONRPC: "2.0",
|
JSONRPC: "2.0",
|
||||||
ID: 1,
|
ID: 1,
|
||||||
Method: "initialize",
|
Method: "initialize",
|
||||||
Params: params,
|
Params: params,
|
||||||
}, true, &payload); err == nil {
|
}, true, &payload)
|
||||||
|
if err == nil {
|
||||||
if payload.ProtocolVersion == "" {
|
if payload.ProtocolVersion == "" {
|
||||||
payload.ProtocolVersion = version
|
payload.ProtocolVersion = version
|
||||||
}
|
}
|
||||||
payload.RequestedProtocolVersion = version
|
payload.RequestedProtocolVersion = version
|
||||||
return payload, nil
|
return payload, nil
|
||||||
}
|
}
|
||||||
|
lastErr = err
|
||||||
|
// Only protocol-level JSON-RPC errors justify trying another version.
|
||||||
|
// Transport/HTTP failures (dial timeout, connection refused, HTTP 5xx,
|
||||||
|
// etc.) fail identically regardless of protocol version, so looping
|
||||||
|
// over three versions only multiplies the dial cost — e.g. on an
|
||||||
|
// unreachable endpoint, three 3-second dials amount to 9 seconds
|
||||||
|
// before the outer context surrenders.
|
||||||
|
var callErr *CallError
|
||||||
|
if !errors.As(err, &callErr) || callErr.Stage != CallStageJSONRPC {
|
||||||
|
return InitializeResult{}, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if lastErr != nil {
|
||||||
|
return InitializeResult{}, lastErr
|
||||||
}
|
}
|
||||||
return InitializeResult{}, apperrors.NewDiscovery(fmt.Sprintf("initialize failed for all supported protocol versions at %s", RedactURL(endpoint)))
|
return InitializeResult{}, apperrors.NewDiscovery(fmt.Sprintf("initialize failed for all supported protocol versions at %s", RedactURL(endpoint)))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
stderrors "errors"
|
stderrors "errors"
|
||||||
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"os"
|
"os"
|
||||||
@@ -71,6 +72,72 @@ func TestInitializeNegotiatesProtocolVersion(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestInitializeShortCircuitsOnHTTPError(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
// Track which protocol versions actually get sent. With the short-circuit
|
||||||
|
// in place, a transport-layer (HTTP) failure should fail Initialize on the
|
||||||
|
// FIRST version without iterating through every supported version. Without
|
||||||
|
// the short-circuit, three round-trips would happen — needlessly tripling
|
||||||
|
// every CLI startup when a plugin endpoint is broken (issue #119).
|
||||||
|
var seenVersions []string
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var req map[string]any
|
||||||
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||||
|
t.Fatalf("decode request: %v", err)
|
||||||
|
}
|
||||||
|
params := req["params"].(map[string]any)
|
||||||
|
seenVersions = append(seenVersions, params["protocolVersion"].(string))
|
||||||
|
http.Error(w, "boom", http.StatusInternalServerError)
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
client := NewClient(server.Client())
|
||||||
|
client.MaxRetries = 0 // skip the HTTP retry loop — we only care about version iteration
|
||||||
|
|
||||||
|
if _, err := client.Initialize(context.Background(), server.URL); err == nil {
|
||||||
|
t.Fatal("Initialize() error = nil, want HTTP error")
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(seenVersions) != 1 {
|
||||||
|
t.Fatalf("Initialize() attempted %d protocol versions (%v), want 1 — HTTP failures must short-circuit",
|
||||||
|
len(seenVersions), seenVersions)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestInitializeShortCircuitsOnDialFailure(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
// Bind to an ephemeral port, then close the listener. Subsequent connects
|
||||||
|
// to that address fail with "connection refused" almost instantly. With
|
||||||
|
// the short-circuit, three protocol versions would otherwise stack three
|
||||||
|
// dial-error returns; we only want one — and we want Initialize to return
|
||||||
|
// well under the per-dial budget.
|
||||||
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("listen: %v", err)
|
||||||
|
}
|
||||||
|
endpoint := "http://" + ln.Addr().String()
|
||||||
|
_ = ln.Close()
|
||||||
|
|
||||||
|
client := NewClient(nil)
|
||||||
|
client.MaxRetries = 0
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
if _, err := client.Initialize(context.Background(), endpoint); err == nil {
|
||||||
|
t.Fatal("Initialize() error = nil, want dial failure")
|
||||||
|
}
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
|
||||||
|
// Three dial attempts (one per supported version) on a refused-connection
|
||||||
|
// path is still fast on loopback, so this assertion is a sanity bound, not
|
||||||
|
// the primary signal — but if the short-circuit regresses, on a real
|
||||||
|
// unreachable address this jumps from one dial timeout to three.
|
||||||
|
if elapsed > 2*time.Second {
|
||||||
|
t.Fatalf("Initialize() took %v, want <2s — dial failure should short-circuit", elapsed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestListToolsRetriesOnServerError(t *testing.T) {
|
func TestListToolsRetriesOnServerError(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user