From 25e129bd2b5c96c24a587b70629e921cf4084d15 Mon Sep 17 00:00:00 2001 From: erezrokah Date: Tue, 1 Sep 2026 17:55:31 +0100 Subject: [PATCH 1/3] fix: Bound plugin downloads with transport timeouts http.DefaultClient has no timeout of any kind, so a plugin asset server that accepts the connection and then goes silent holds the attempt until a middlebox tears the connection down minutes later. Route both asset requests through a dedicated client with dial, TLS handshake, response-header and per-read idle deadlines. Timeouts produced by that client are re-labelled as retryable, since they otherwise reach the classifier indistinguishable from the caller's own deadline. --- managedplugin/download.go | 13 +- managedplugin/download_client.go | 66 ++++++++++ managedplugin/download_client_test.go | 180 ++++++++++++++++++++++++++ managedplugin/download_retry.go | 25 +++- managedplugin/download_retry_test.go | 1 + 5 files changed, 280 insertions(+), 5 deletions(-) create mode 100644 managedplugin/download_client.go create mode 100644 managedplugin/download_client_test.go diff --git a/managedplugin/download.go b/managedplugin/download.go index f5eb9e7..5bc1bf4 100644 --- a/managedplugin/download.go +++ b/managedplugin/download.go @@ -93,8 +93,11 @@ func getURLLocation(ctx context.Context, org string, name string, version string if err != nil { return fmt.Errorf("failed create request %s: %w", urlForLog, redactURLError(err)) } - resp, err := http.DefaultClient.Do(req) + resp, err := downloadClient.Do(req) if err != nil { + if terr := downloadTimeoutError(ctx, urlForLog, err); terr != nil { + return terr + } return fmt.Errorf("failed to get url %s: %w", urlForLog, redactURLError(err)) } resp.Body.Close() @@ -370,8 +373,11 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop } // Do http request - resp, err := http.DefaultClient.Do(req) + resp, err := downloadClient.Do(req) if err != nil { + if terr := downloadTimeoutError(ctx, urlForLog, err); terr != nil { + return terr + } return fmt.Errorf("failed to get url %s: %w", urlForLog, redactURLError(err)) } defer resp.Body.Close() @@ -400,6 +406,9 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop // Write the body to file written, err := io.Copy(io.MultiWriter(writers...), resp.Body) if err != nil { + if terr := downloadTimeoutError(ctx, urlForLog, err); terr != nil { + return terr + } return fmt.Errorf("failed to copy body to file %s: %w", out.Name(), err) } if resp.ContentLength >= 0 && written != resp.ContentLength { diff --git a/managedplugin/download_client.go b/managedplugin/download_client.go new file mode 100644 index 0000000..9f2031c --- /dev/null +++ b/managedplugin/download_client.go @@ -0,0 +1,66 @@ +package managedplugin + +import ( + "context" + "net" + "net/http" + "time" +) + +// Overridable so tests do not pay the real timeouts. +var ( + downloadDialTimeout = 10 * time.Second + downloadTLSHandshakeTimeout = 10 * time.Second + downloadResponseHeaderTimeout = 30 * time.Second + downloadIdleReadTimeout = 30 * time.Second +) + +var downloadClient = newDownloadClient() + +// idleTimeoutConn arms a read deadline immediately before every read, so a peer +// that stops sending fails the connection instead of holding it until some +// middlebox tears it down minutes later. Arming per read rather than once means +// a slow but moving transfer never trips, and a caller stalled in its own write +// path is not blamed on the server. +type idleTimeoutConn struct { + net.Conn + idle time.Duration +} + +func (c *idleTimeoutConn) Read(b []byte) (int, error) { + if err := c.SetReadDeadline(time.Now().Add(c.idle)); err != nil { + return 0, err + } + return c.Conn.Read(b) +} + +// newDownloadClient builds the client used for every plugin asset request. +// http.DefaultClient has no timeout of any kind, so a silent server hangs a +// download until the connection is torn down externally. +// +// ForceAttemptHTTP2 is deliberately left off: a read deadline is per connection, +// and HTTP/2 multiplexes streams onto one connection, so an idle deadline there +// would be shared across concurrent requests. +func newDownloadClient() *http.Client { + dialer := &net.Dialer{ + Timeout: downloadDialTimeout, + KeepAlive: 30 * time.Second, + } + idle := downloadIdleReadTimeout + return &http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + conn, err := dialer.DialContext(ctx, network, addr) + if err != nil { + return nil, err + } + return &idleTimeoutConn{Conn: conn, idle: idle}, nil + }, + TLSHandshakeTimeout: downloadTLSHandshakeTimeout, + ResponseHeaderTimeout: downloadResponseHeaderTimeout, + IdleConnTimeout: downloadIdleReadTimeout, + MaxIdleConnsPerHost: http.DefaultMaxIdleConnsPerHost, + }, + } +} diff --git a/managedplugin/download_client_test.go b/managedplugin/download_client_test.go new file mode 100644 index 0000000..ff100a9 --- /dev/null +++ b/managedplugin/download_client_test.go @@ -0,0 +1,180 @@ +package managedplugin + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// fastStall shrinks the transport timeouts so the stall tests do not wait the +// real 30s per attempt. +func fastStall(t *testing.T, d time.Duration) { + t.Helper() + + prevHeader, prevIdle, prevClient := downloadResponseHeaderTimeout, downloadIdleReadTimeout, downloadClient + downloadResponseHeaderTimeout, downloadIdleReadTimeout = d, d + downloadClient = newDownloadClient() + t.Cleanup(func() { + downloadResponseHeaderTimeout, downloadIdleReadTimeout = prevHeader, prevIdle + downloadClient = prevClient + }) +} + +// TestDownloadFileRetriesStalledHeaders covers a server that accepts the +// connection and then never responds. Without ResponseHeaderTimeout the attempt +// hangs until the connection is torn down externally. +func TestDownloadFileRetriesStalledHeaders(t *testing.T) { + fastRetries(t) + fastStall(t, 150*time.Millisecond) + + body := []byte("payload") + release := make(chan struct{}) + + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if attempts.Add(1) == 1 { + <-release + return + } + _, _ = w.Write(body) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(release) }) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + checksum, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.NoError(t, err) + require.GreaterOrEqual(t, attempts.Load(), int32(2)) + require.Equal(t, sha256Hex(body), checksum) +} + +// TestDownloadFileRetriesStalledBody covers a server that sends part of the body +// and then goes silent without closing the connection. +func TestDownloadFileRetriesStalledBody(t *testing.T) { + fastRetries(t) + fastStall(t, 150*time.Millisecond) + + body := []byte("cloudquery-plugin-binary-payload") + release := make(chan struct{}) + + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if attempts.Add(1) == 1 { + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body))) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(body[:5]) + w.(http.Flusher).Flush() + <-release + return + } + _, _ = w.Write(body) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(release) }) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + checksum, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.NoError(t, err) + require.GreaterOrEqual(t, attempts.Load(), int32(2)) + + written, err := os.ReadFile(localPath) + require.NoError(t, err) + require.Equal(t, body, written) + require.Equal(t, sha256Hex(body), checksum) +} + +// TestDownloadFileSlowButProgressingSucceeds pins the boundary the idle deadline +// must not cross: a transfer whose total duration exceeds the timeout still +// succeeds in one attempt as long as bytes keep arriving. +func TestDownloadFileSlowButProgressingSucceeds(t *testing.T) { + fastRetries(t) + fastStall(t, 300*time.Millisecond) + + body := []byte("slow-but-moving-plugin-payload") + + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body))) + w.WriteHeader(http.StatusOK) + for _, chunk := range [][]byte{body[:6], body[6:12], body[12:18], body[18:24], body[24:]} { + _, _ = w.Write(chunk) + w.(http.Flusher).Flush() + time.Sleep(50 * time.Millisecond) + } + })) + t.Cleanup(server.Close) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + checksum, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.NoError(t, err) + require.EqualValues(t, 1, attempts.Load(), "a transfer that keeps making progress must not trip the idle deadline") + require.Equal(t, sha256Hex(body), checksum) +} + +// TestDownloadFileCallerCancelIsNotRetried pins the boundary between the idle +// deadline (retryable) and the caller's own cancellation (terminal). +func TestDownloadFileCallerCancelIsNotRetried(t *testing.T) { + fastRetries(t) + + release := make(chan struct{}) + + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + w.Header().Set("Content-Length", "32") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("early")) + w.(http.Flusher).Flush() + <-release + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(release) }) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(100 * time.Millisecond) + cancel() + }() + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + _, err := downloadFile(ctx, localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.Error(t, err) + require.ErrorIs(t, err, context.Canceled) + require.False(t, IsTransientDownloadError(err), "the caller's own cancellation must not be retried") +} + +// TestDownloadFileCallerDeadlineIsNotRetried guards the distinction the +// transport timeouts blur: our own timeout and the caller's deadline both +// satisfy errors.Is(err, context.DeadlineExceeded), and only ours is retryable. +func TestDownloadFileCallerDeadlineIsNotRetried(t *testing.T) { + fastRetries(t) + fastStall(t, 10*time.Second) + + release := make(chan struct{}) + + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + <-release + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(release) }) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + _, err := downloadFile(ctx, localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.Error(t, err) + require.NotErrorIs(t, err, errDownloadStalled) + require.False(t, IsTransientDownloadError(err), "the caller's own deadline must not be retried") +} diff --git a/managedplugin/download_retry.go b/managedplugin/download_retry.go index 056186f..2cd9a32 100644 --- a/managedplugin/download_retry.go +++ b/managedplugin/download_retry.go @@ -13,8 +13,9 @@ import ( ) var ( - errNotFound = errors.New("not found") - errShortRead = errors.New("truncated response body") + errNotFound = errors.New("not found") + errShortRead = errors.New("truncated response body") + errDownloadStalled = errors.New("download stalled") ) // Overridable so tests do not pay the real backoff. @@ -81,7 +82,8 @@ func isRetryableDownloadError(err error) bool { } switch { - case errors.Is(err, errShortRead), + case errors.Is(err, errDownloadStalled), + errors.Is(err, errShortRead), errors.Is(err, io.ErrUnexpectedEOF), errors.Is(err, io.EOF), errors.Is(err, syscall.ECONNRESET), @@ -114,6 +116,23 @@ func isRetryableDownloadError(err error) bool { return false } +// downloadTimeoutError re-labels a timeout produced by our own transport, which +// otherwise reaches the classifier indistinguishable from the caller's deadline: +// both satisfy errors.Is(err, context.DeadlineExceeded), and only the caller's is +// terminal. The underlying error is formatted with %v so that shared +// context.DeadlineExceeded does not travel on in the chain. Returns nil when the +// caller's own context ended the request, or when err is not a timeout at all. +func downloadTimeoutError(ctx context.Context, urlForLog string, err error) error { + if ctx.Err() != nil { + return nil + } + var netErr net.Error + if !errors.As(err, &netErr) || !netErr.Timeout() { + return nil + } + return fmt.Errorf("%w: no data from %s: %v", errDownloadStalled, urlForLog, redactURLError(err)) +} + // redactURLQuery strips the query string so that the signed download token never // reaches stdout or a log aggregator. func redactURLQuery(rawURL string) string { diff --git a/managedplugin/download_retry_test.go b/managedplugin/download_retry_test.go index ba62afd..ddc19dc 100644 --- a/managedplugin/download_retry_test.go +++ b/managedplugin/download_retry_test.go @@ -58,6 +58,7 @@ func TestIsRetryableDownloadError(t *testing.T) { {name: "bad gateway 502", err: &httpStatusError{statusCode: http.StatusBadGateway}, want: true}, {name: "service unavailable 503", err: &httpStatusError{statusCode: http.StatusServiceUnavailable}, want: true}, + {name: "stalled download", err: fmt.Errorf("%w: no data from host: i/o timeout", errDownloadStalled), want: true}, {name: "context canceled", err: fmt.Errorf("get url: %w", context.Canceled), want: false}, {name: "context deadline exceeded", err: fmt.Errorf("get url: %w", context.DeadlineExceeded), want: false}, {name: "checksum mismatch is permanent", err: errors.New("checksum mismatch: expected abc, got def"), want: false}, From b7183cb76a4ae605b90256b24ae4f4cbf1640d0e Mon Sep 17 00:00:00 2001 From: erezrokah Date: Tue, 1 Sep 2026 17:56:38 +0100 Subject: [PATCH 2/3] chore: Trim comments --- managedplugin/download_client.go | 16 +++++----------- managedplugin/download_client_test.go | 16 +--------------- managedplugin/download_retry.go | 11 +++++------ 3 files changed, 11 insertions(+), 32 deletions(-) diff --git a/managedplugin/download_client.go b/managedplugin/download_client.go index 9f2031c..8b4373f 100644 --- a/managedplugin/download_client.go +++ b/managedplugin/download_client.go @@ -17,11 +17,9 @@ var ( var downloadClient = newDownloadClient() -// idleTimeoutConn arms a read deadline immediately before every read, so a peer -// that stops sending fails the connection instead of holding it until some -// middlebox tears it down minutes later. Arming per read rather than once means -// a slow but moving transfer never trips, and a caller stalled in its own write -// path is not blamed on the server. +// idleTimeoutConn arms the deadline per read, not once: a slow but moving +// transfer never trips, and a caller stalled in its own write path is not +// blamed on the server. type idleTimeoutConn struct { net.Conn idle time.Duration @@ -34,13 +32,9 @@ func (c *idleTimeoutConn) Read(b []byte) (int, error) { return c.Conn.Read(b) } -// newDownloadClient builds the client used for every plugin asset request. -// http.DefaultClient has no timeout of any kind, so a silent server hangs a -// download until the connection is torn down externally. -// // ForceAttemptHTTP2 is deliberately left off: a read deadline is per connection, -// and HTTP/2 multiplexes streams onto one connection, so an idle deadline there -// would be shared across concurrent requests. +// and HTTP/2 would multiplex streams onto one, sharing the idle deadline across +// concurrent requests. func newDownloadClient() *http.Client { dialer := &net.Dialer{ Timeout: downloadDialTimeout, diff --git a/managedplugin/download_client_test.go b/managedplugin/download_client_test.go index ff100a9..b163656 100644 --- a/managedplugin/download_client_test.go +++ b/managedplugin/download_client_test.go @@ -14,8 +14,7 @@ import ( "github.com/stretchr/testify/require" ) -// fastStall shrinks the transport timeouts so the stall tests do not wait the -// real 30s per attempt. +// fastStall shrinks the transport timeouts so the tests do not wait the real 30s. func fastStall(t *testing.T, d time.Duration) { t.Helper() @@ -28,9 +27,6 @@ func fastStall(t *testing.T, d time.Duration) { }) } -// TestDownloadFileRetriesStalledHeaders covers a server that accepts the -// connection and then never responds. Without ResponseHeaderTimeout the attempt -// hangs until the connection is torn down externally. func TestDownloadFileRetriesStalledHeaders(t *testing.T) { fastRetries(t) fastStall(t, 150*time.Millisecond) @@ -56,8 +52,6 @@ func TestDownloadFileRetriesStalledHeaders(t *testing.T) { require.Equal(t, sha256Hex(body), checksum) } -// TestDownloadFileRetriesStalledBody covers a server that sends part of the body -// and then goes silent without closing the connection. func TestDownloadFileRetriesStalledBody(t *testing.T) { fastRetries(t) fastStall(t, 150*time.Millisecond) @@ -91,9 +85,6 @@ func TestDownloadFileRetriesStalledBody(t *testing.T) { require.Equal(t, sha256Hex(body), checksum) } -// TestDownloadFileSlowButProgressingSucceeds pins the boundary the idle deadline -// must not cross: a transfer whose total duration exceeds the timeout still -// succeeds in one attempt as long as bytes keep arriving. func TestDownloadFileSlowButProgressingSucceeds(t *testing.T) { fastRetries(t) fastStall(t, 300*time.Millisecond) @@ -120,8 +111,6 @@ func TestDownloadFileSlowButProgressingSucceeds(t *testing.T) { require.Equal(t, sha256Hex(body), checksum) } -// TestDownloadFileCallerCancelIsNotRetried pins the boundary between the idle -// deadline (retryable) and the caller's own cancellation (terminal). func TestDownloadFileCallerCancelIsNotRetried(t *testing.T) { fastRetries(t) @@ -152,9 +141,6 @@ func TestDownloadFileCallerCancelIsNotRetried(t *testing.T) { require.False(t, IsTransientDownloadError(err), "the caller's own cancellation must not be retried") } -// TestDownloadFileCallerDeadlineIsNotRetried guards the distinction the -// transport timeouts blur: our own timeout and the caller's deadline both -// satisfy errors.Is(err, context.DeadlineExceeded), and only ours is retryable. func TestDownloadFileCallerDeadlineIsNotRetried(t *testing.T) { fastRetries(t) fastStall(t, 10*time.Second) diff --git a/managedplugin/download_retry.go b/managedplugin/download_retry.go index 2cd9a32..1f94c51 100644 --- a/managedplugin/download_retry.go +++ b/managedplugin/download_retry.go @@ -116,12 +116,11 @@ func isRetryableDownloadError(err error) bool { return false } -// downloadTimeoutError re-labels a timeout produced by our own transport, which -// otherwise reaches the classifier indistinguishable from the caller's deadline: -// both satisfy errors.Is(err, context.DeadlineExceeded), and only the caller's is -// terminal. The underlying error is formatted with %v so that shared -// context.DeadlineExceeded does not travel on in the chain. Returns nil when the -// caller's own context ended the request, or when err is not a timeout at all. +// downloadTimeoutError re-labels a timeout from our own transport, which reaches +// the classifier indistinguishable from the caller's deadline: both satisfy +// errors.Is(err, context.DeadlineExceeded), and only the caller's is terminal. +// The cause is formatted with %v so that shared error does not travel on in the +// chain. func downloadTimeoutError(ctx context.Context, urlForLog string, err error) error { if ctx.Err() != nil { return nil From a93aaebee5262e6cdd3f0700435094a83dbbc432 Mon Sep 17 00:00:00 2001 From: erezrokah Date: Wed, 2 Sep 2026 09:37:54 +0100 Subject: [PATCH 3/3] chore: Strip code comments --- managedplugin/download_client.go | 7 ------- managedplugin/download_client_test.go | 1 - managedplugin/download_retry.go | 5 ----- 3 files changed, 13 deletions(-) diff --git a/managedplugin/download_client.go b/managedplugin/download_client.go index 8b4373f..c071934 100644 --- a/managedplugin/download_client.go +++ b/managedplugin/download_client.go @@ -7,7 +7,6 @@ import ( "time" ) -// Overridable so tests do not pay the real timeouts. var ( downloadDialTimeout = 10 * time.Second downloadTLSHandshakeTimeout = 10 * time.Second @@ -17,9 +16,6 @@ var ( var downloadClient = newDownloadClient() -// idleTimeoutConn arms the deadline per read, not once: a slow but moving -// transfer never trips, and a caller stalled in its own write path is not -// blamed on the server. type idleTimeoutConn struct { net.Conn idle time.Duration @@ -32,9 +28,6 @@ func (c *idleTimeoutConn) Read(b []byte) (int, error) { return c.Conn.Read(b) } -// ForceAttemptHTTP2 is deliberately left off: a read deadline is per connection, -// and HTTP/2 would multiplex streams onto one, sharing the idle deadline across -// concurrent requests. func newDownloadClient() *http.Client { dialer := &net.Dialer{ Timeout: downloadDialTimeout, diff --git a/managedplugin/download_client_test.go b/managedplugin/download_client_test.go index b163656..950057f 100644 --- a/managedplugin/download_client_test.go +++ b/managedplugin/download_client_test.go @@ -14,7 +14,6 @@ import ( "github.com/stretchr/testify/require" ) -// fastStall shrinks the transport timeouts so the tests do not wait the real 30s. func fastStall(t *testing.T, d time.Duration) { t.Helper() diff --git a/managedplugin/download_retry.go b/managedplugin/download_retry.go index 1f94c51..849cead 100644 --- a/managedplugin/download_retry.go +++ b/managedplugin/download_retry.go @@ -116,11 +116,6 @@ func isRetryableDownloadError(err error) bool { return false } -// downloadTimeoutError re-labels a timeout from our own transport, which reaches -// the classifier indistinguishable from the caller's deadline: both satisfy -// errors.Is(err, context.DeadlineExceeded), and only the caller's is terminal. -// The cause is formatted with %v so that shared error does not travel on in the -// chain. func downloadTimeoutError(ctx context.Context, urlForLog string, err error) error { if ctx.Err() != nil { return nil