From ddc1cef3cfaa20e735cac2058986553eb7ff61be Mon Sep 17 00:00:00 2001 From: Samuele Verzi Date: Thu, 6 Aug 2026 11:17:29 +0200 Subject: [PATCH] Stop severing plugin artifact pulls at fixed timeouts thv ai-plugin had both ceilings that #6212 fixed for skills, for the same reason: its client is a near-copy of the skills client and its router sat among the standard routers. The client gave up after 30s and blamed server availability, when the server was healthy and mid-pull. Past that, the plugins router inherited the flat 60s cap that the workload and skills routers are both exempt from, on the grounds that artifact pulls take minutes. - classify timeouts as ErrRequestTimeout, separate from unreachability, and leave caller cancellation as neither - raise the client default and allow TOOLHIVE_API_TIMEOUT to override it - give the plugins router per-route timeouts, long ones on install, build, and push Follows #6212. --- cmd/thv/app/ai_plugin_helpers.go | 16 +++- pkg/api/server.go | 4 +- pkg/api/v1/plugins.go | 28 +++--- pkg/plugins/client/client.go | 79 ++++++++++++++--- pkg/plugins/client/client_test.go | 140 ++++++++++++++++++++++++++++++ 5 files changed, 242 insertions(+), 25 deletions(-) diff --git a/cmd/thv/app/ai_plugin_helpers.go b/cmd/thv/app/ai_plugin_helpers.go index fd90be13f9..78f7b9677d 100644 --- a/cmd/thv/app/ai_plugin_helpers.go +++ b/cmd/thv/app/ai_plugin_helpers.go @@ -39,13 +39,21 @@ func completeAIPluginNames(cmd *cobra.Command, args []string, _ string) ([]strin return names, cobra.ShellCompDirectiveNoFileComp } -// formatAIPluginError wraps an error with contextual information. If the -// underlying cause is ErrServerUnreachable it appends a helpful hint. +// formatAIPluginError wraps an error with contextual information, appending a +// hint that matches the actual failure — a timed-out request and an absent +// server need different advice. func formatAIPluginError(action string, err error) error { - if errors.Is(err, pluginclient.ErrServerUnreachable) { + switch { + case errors.Is(err, pluginclient.ErrRequestTimeout): + return fmt.Errorf( + "failed to %s: %w\nHint: the server is running and was still working; "+ + "raise the limit with TOOLHIVE_API_TIMEOUT (e.g. TOOLHIVE_API_TIMEOUT=30m)", + action, err) + case errors.Is(err, pluginclient.ErrServerUnreachable): return fmt.Errorf("failed to %s: %w\nHint: ensure 'thv serve' is running", action, err) + default: + return fmt.Errorf("failed to %s: %w", action, err) } - return fmt.Errorf("failed to %s: %w", action, err) } // validateAIPluginScope returns a PreRunE that validates the --scope flag. diff --git a/pkg/api/server.go b/pkg/api/server.go index 5226237591..d370a48f01 100644 --- a/pkg/api/server.go +++ b/pkg/api/server.go @@ -413,6 +413,9 @@ func (b *ServerBuilder) setupDefaultRoutes(r *chi.Mux) { // artifacts, so a flat 60s cap would sever them mid-transfer. r.Mount("/api/v1beta/skills", v1.SkillsRouter(b.skillManager)) + // Plugins router likewise: install, build, and push move OCI artifacts. + r.Mount("/api/v1beta/plugins", v1.PluginsRouter(b.pluginManager)) + // All other routes get standard timeout standardRouters := map[string]http.Handler{ "/health": v1.HealthcheckRouter(b.containerRuntime, b.nonce), @@ -422,7 +425,6 @@ func (b *ServerBuilder) setupDefaultRoutes(r *chi.Mux) { "/api/v1beta/clients": v1.ClientRouter(b.clientManager, b.workloadManager, b.groupManager), "/api/v1beta/secrets": v1.SecretsRouter(), "/api/v1beta/groups": v1.GroupsRouter(b.groupManager, b.workloadManager, b.clientManager), - "/api/v1beta/plugins": v1.PluginsRouter(b.pluginManager), "/registry": v1.RegistryV01Router(), } for prefix, router := range standardRouters { diff --git a/pkg/api/v1/plugins.go b/pkg/api/v1/plugins.go index 73da544f83..2ff6aee30e 100644 --- a/pkg/api/v1/plugins.go +++ b/pkg/api/v1/plugins.go @@ -9,6 +9,7 @@ import ( "net/http" "github.com/go-chi/chi/v5" + "github.com/go-chi/chi/v5/middleware" "github.com/stacklok/toolhive-core/httperr" apierrors "github.com/stacklok/toolhive/pkg/api/errors" @@ -26,17 +27,24 @@ func PluginsRouter(pluginService plugins.PluginService) http.Handler { pluginService: pluginService, } + // Mirrors WorkloadRouter and SkillsRouter: routes that move OCI artifacts + // get a timeout sized for the transfer, everything else keeps the short + // one. Without the split these routes inherit the flat 60s applied to + // standard routers, which severs a cold artifact pull mid-flight. + stdTimeout := middleware.Timeout(standardRouteTimeout) + longTimeout := middleware.Timeout(longRunningRouteTimeout) + r := chi.NewRouter() - r.Get("/", apierrors.ErrorHandler(routes.listPlugins)) - r.Post("/", apierrors.ErrorHandler(routes.installPlugin)) - r.Delete("/{name}", apierrors.ErrorHandler(routes.uninstallPlugin)) - r.Get("/{name}", apierrors.ErrorHandler(routes.getPluginInfo)) - r.Post("/validate", apierrors.ErrorHandler(routes.validatePlugin)) - r.Post("/build", apierrors.ErrorHandler(routes.buildPlugin)) - r.Post("/push", apierrors.ErrorHandler(routes.pushPlugin)) - r.Get("/builds", apierrors.ErrorHandler(routes.listBuilds)) - r.Delete("/builds/{tag}", apierrors.ErrorHandler(routes.deleteBuild)) - r.Get("/content", apierrors.ErrorHandler(routes.getPluginContent)) + r.With(stdTimeout).Get("/", apierrors.ErrorHandler(routes.listPlugins)) + r.With(longTimeout).Post("/", apierrors.ErrorHandler(routes.installPlugin)) + r.With(stdTimeout).Delete("/{name}", apierrors.ErrorHandler(routes.uninstallPlugin)) + r.With(stdTimeout).Get("/{name}", apierrors.ErrorHandler(routes.getPluginInfo)) + r.With(stdTimeout).Post("/validate", apierrors.ErrorHandler(routes.validatePlugin)) + r.With(longTimeout).Post("/build", apierrors.ErrorHandler(routes.buildPlugin)) + r.With(longTimeout).Post("/push", apierrors.ErrorHandler(routes.pushPlugin)) + r.With(stdTimeout).Get("/builds", apierrors.ErrorHandler(routes.listBuilds)) + r.With(stdTimeout).Delete("/builds/{tag}", apierrors.ErrorHandler(routes.deleteBuild)) + r.With(stdTimeout).Get("/content", apierrors.ErrorHandler(routes.getPluginContent)) return r } diff --git a/pkg/plugins/client/client.go b/pkg/plugins/client/client.go index 264170a51e..cb75e1c04c 100644 --- a/pkg/plugins/client/client.go +++ b/pkg/plugins/client/client.go @@ -12,6 +12,7 @@ import ( "fmt" "io" "log/slog" + "net" "net/http" "net/url" "strings" @@ -24,10 +25,18 @@ import ( ) const ( - pluginsBasePath = "/api/v1beta/plugins" - defaultBaseURL = "http://127.0.0.1:8080" - defaultTimeout = 30 * time.Second + pluginsBasePath = "/api/v1beta/plugins" + defaultBaseURL = "http://127.0.0.1:8080" + // defaultTimeout is a backstop against a wedged server, not a budget for + // the operation. Install, build, and push move OCI artifacts of unbounded + // size over the user's connection, so a timeout tight enough to feel + // responsive would abort legitimate work — and because giving up cancels + // the request context, it aborts it server-side too, leaving nothing for + // a retry to reuse. Callers who want to fail fast can set + // TOOLHIVE_API_TIMEOUT or pass WithTimeout. + defaultTimeout = 10 * time.Minute envAPIURL = "TOOLHIVE_API_URL" + envAPITimeout = "TOOLHIVE_API_TIMEOUT" maxResponseSize = 1 << 20 // 1 MiB — defensive response cap; consistent with the skills client maxErrorBodySize = 1 << 16 // 64 KiB — matches auth/token and DCR limits ) @@ -37,6 +46,13 @@ const ( // running. var ErrServerUnreachable = errors.New("could not reach ToolHive API server — is 'thv serve' running?") +// ErrRequestTimeout is returned when the server was reached but did not +// respond within the client timeout. This is a distinct condition from +// ErrServerUnreachable: the server is healthy and was working on the request. +// Note that abandoning the request cancels its context server-side, so the +// operation does not continue in the background and retrying restarts it. +var ErrRequestTimeout = errors.New("the ToolHive API server did not respond within the client timeout") + // Compile-time interface check. var _ plugins.PluginService = (*Client)(nil) @@ -96,23 +112,47 @@ type discoverFunc func(ctx context.Context) (string, []Option) // envReader and discover dependencies are injected so each resolution step // can be exercised in isolation. func newDefaultClientWithEnv(ctx context.Context, envReader env.Reader, discover discoverFunc, opts ...Option) *Client { + // An env-supplied timeout applies to every resolution path, but ranks + // below caller-supplied options so a WithTimeout argument still wins. + withEnvTimeout := func(base []Option) []Option { + merged := make([]Option, 0, len(base)+len(opts)+1) + merged = append(merged, base...) + if d, ok := timeoutFromEnv(envReader); ok { + merged = append(merged, WithTimeout(d)) + } + return append(merged, opts...) + } + // 1. Explicit env var override always wins. if base := envReader.Getenv(envAPIURL); base != "" { - return NewClient(base, opts...) + return NewClient(base, withEnvTimeout(nil)...) } // 2. Try server discovery. if base, httpOpts := discover(ctx); base != "" { // Discovery opts go first so caller-supplied opts can override them // (e.g. a caller-provided WithTimeout replaces the discovery default). - merged := make([]Option, 0, len(httpOpts)+len(opts)) - merged = append(merged, httpOpts...) - merged = append(merged, opts...) - return NewClient(base, merged...) + return NewClient(base, withEnvTimeout(httpOpts)...) } // 3. Fall back to the default URL. - return NewClient(defaultBaseURL, opts...) + return NewClient(defaultBaseURL, withEnvTimeout(nil)...) +} + +// timeoutFromEnv reads TOOLHIVE_API_TIMEOUT as a Go duration (e.g. "45s", +// "5m"). An unset, unparsable, or non-positive value is ignored so a typo +// degrades to the default rather than disabling the timeout entirely. +func timeoutFromEnv(envReader env.Reader) (time.Duration, bool) { + raw := strings.TrimSpace(envReader.Getenv(envAPITimeout)) + if raw == "" { + return 0, false + } + d, err := time.ParseDuration(raw) + if err != nil || d <= 0 { + slog.Warn("ignoring invalid "+envAPITimeout, "value", raw, "error", err) + return 0, false + } + return d, true } // resolveViaDiscovery attempts to find a running server via the discovery file. @@ -313,7 +353,7 @@ func (c *Client) doJSONRequest( resp, err := c.httpClient.Do(req) // #nosec G704 -- baseURL is a trusted local API server URL if err != nil { - return fmt.Errorf("%w: %w", ErrServerUnreachable, err) + return fmt.Errorf("%w: %w", classifyTransportError(ctx, err), err) } defer func() { _ = resp.Body.Close() }() @@ -333,6 +373,25 @@ func (c *Client) doJSONRequest( return nil } +// classifyTransportError decides which sentinel a failed round-trip belongs +// to. Reporting a timeout as an unreachable server sends users to check +// whether "thv serve" is running when it is running and was mid-operation. +func classifyTransportError(ctx context.Context, err error) error { + // The caller's own context ending takes precedence — neither the server's + // availability nor the client timeout is why the request stopped. + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return ErrRequestTimeout + } + if errors.Is(err, context.DeadlineExceeded) { + return ErrRequestTimeout + } + return ErrServerUnreachable +} + // handleErrorResponse reads the response body and returns an *httperr.CodedError. func handleErrorResponse(resp *http.Response) error { body, err := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodySize)) diff --git a/pkg/plugins/client/client_test.go b/pkg/plugins/client/client_test.go index afd6dfce72..e87a11f524 100644 --- a/pkg/plugins/client/client_test.go +++ b/pkg/plugins/client/client_test.go @@ -8,6 +8,7 @@ import ( "encoding/json" "errors" "io" + "net" "net/http" "net/http/httptest" "testing" @@ -775,6 +776,7 @@ func TestNewDefaultClient(t *testing.T) { ctrl := gomock.NewController(t) mockEnv := envmocks.NewMockReader(ctrl) mockEnv.EXPECT().Getenv(envAPIURL).Return("") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("").AnyTimes() c := newDefaultClientWithEnv(t.Context(), mockEnv, noDiscovery) assert.Equal(t, defaultBaseURL, c.baseURL) @@ -785,6 +787,7 @@ func TestNewDefaultClient(t *testing.T) { ctrl := gomock.NewController(t) mockEnv := envmocks.NewMockReader(ctrl) mockEnv.EXPECT().Getenv(envAPIURL).Return("http://localhost:9999") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("").AnyTimes() c := newDefaultClientWithEnv(t.Context(), mockEnv, failDiscovery(t)) assert.Equal(t, "http://localhost:9999", c.baseURL) @@ -795,6 +798,7 @@ func TestNewDefaultClient(t *testing.T) { ctrl := gomock.NewController(t) mockEnv := envmocks.NewMockReader(ctrl) mockEnv.EXPECT().Getenv(envAPIURL).Return("") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("").AnyTimes() discover := func(context.Context) (string, []Option) { return "http://127.0.0.1:54321", nil @@ -808,6 +812,7 @@ func TestNewDefaultClient(t *testing.T) { ctrl := gomock.NewController(t) mockEnv := envmocks.NewMockReader(ctrl) mockEnv.EXPECT().Getenv(envAPIURL).Return("") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("").AnyTimes() c := newDefaultClientWithEnv(t.Context(), mockEnv, noDiscovery, WithTimeout(5*time.Second)) assert.Equal(t, 5*time.Second, c.httpClient.Timeout) @@ -859,3 +864,138 @@ type failReader struct{} func (*failReader) Read([]byte) (int, error) { return 0, errors.New("simulated read error") } + +func TestTimeoutFromEnv(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + value string + want time.Duration + ok bool + }{ + {name: "unset falls back to the default", value: "", want: 0, ok: false}, + {name: "a duration is honored", value: "45s", want: 45 * time.Second, ok: true}, + {name: "minutes parse", value: "5m", want: 5 * time.Minute, ok: true}, + {name: "surrounding whitespace is tolerated", value: " 90s ", want: 90 * time.Second, ok: true}, + {name: "a bare number is not a duration", value: "60", want: 0, ok: false}, + {name: "garbage degrades to the default", value: "soon", want: 0, ok: false}, + {name: "zero would disable the timeout, so it is ignored", value: "0s", want: 0, ok: false}, + {name: "negative is ignored", value: "-5s", want: 0, ok: false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + mockEnv := envmocks.NewMockReader(ctrl) + mockEnv.EXPECT().Getenv(envAPITimeout).Return(tc.value) + + got, ok := timeoutFromEnv(mockEnv) + assert.Equal(t, tc.ok, ok) + assert.Equal(t, tc.want, got) + }) + } +} + +func TestNewDefaultClientTimeoutPrecedence(t *testing.T) { + t.Parallel() + + noDiscovery := func(context.Context) (string, []Option) { return "", nil } + + t.Run("defaults when the env is unset", func(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + mockEnv := envmocks.NewMockReader(ctrl) + mockEnv.EXPECT().Getenv(envAPIURL).Return("") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("") + + c := newDefaultClientWithEnv(t.Context(), mockEnv, noDiscovery) + assert.Equal(t, defaultTimeout, c.httpClient.Timeout) + }) + + t.Run("env overrides the default", func(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + mockEnv := envmocks.NewMockReader(ctrl) + mockEnv.EXPECT().Getenv(envAPIURL).Return("") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("2m") + + c := newDefaultClientWithEnv(t.Context(), mockEnv, noDiscovery) + assert.Equal(t, 2*time.Minute, c.httpClient.Timeout) + }) + + t.Run("an explicit option outranks the env", func(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + mockEnv := envmocks.NewMockReader(ctrl) + mockEnv.EXPECT().Getenv(envAPIURL).Return("") + mockEnv.EXPECT().Getenv(envAPITimeout).Return("2m") + + c := newDefaultClientWithEnv(t.Context(), mockEnv, noDiscovery, WithTimeout(7*time.Second)) + assert.Equal(t, 7*time.Second, c.httpClient.Timeout) + }) +} + +// TestTimeoutIsReportedAsTimeoutNotUnreachable pins the distinction the CLI +// hint depends on: a healthy-but-slow server must not be reported as absent. +func TestTimeoutIsReportedAsTimeoutNotUnreachable(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + select { + case <-r.Context().Done(): + case <-time.After(30 * time.Second): + } + })) + t.Cleanup(srv.Close) + + c := NewClient(srv.URL, WithTimeout(50*time.Millisecond)) + _, err := c.List(t.Context(), plugins.ListOptions{}) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrRequestTimeout) + assert.NotErrorIs(t, err, ErrServerUnreachable, + "the server answered the connection; only the response was slow") +} + +func TestUnreachableServerIsStillUnreachable(t *testing.T) { + t.Parallel() + + // Bind and immediately release a port so nothing is listening on it. + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := l.Addr().String() + require.NoError(t, l.Close()) + + c := NewClient("http://"+addr, WithTimeout(2*time.Second)) + _, err = c.List(t.Context(), plugins.ListOptions{}) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrServerUnreachable) + assert.NotErrorIs(t, err, ErrRequestTimeout) +} + +// TestCallerCancellationIsNeitherSentinel keeps a user pressing Ctrl-C from +// being reported as a server problem. +func TestCallerCancellationIsNeitherSentinel(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + })) + t.Cleanup(srv.Close) + + ctx, cancel := context.WithCancel(t.Context()) + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + c := NewClient(srv.URL, WithTimeout(30*time.Second)) + _, err := c.List(ctx, plugins.ListOptions{}) + + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled) + assert.NotErrorIs(t, err, ErrRequestTimeout) + assert.NotErrorIs(t, err, ErrServerUnreachable) +}