fix: Address the review on the plugin download retry

Redact the URL that url.Error prints verbatim, so wrapping a transport
error no longer re-exposes the signed token. Retry connection refused,
unreachable networks and DNS failures, and match the Windows WSA socket
messages that the synthetic syscall.E* constants never match. Migrate
the getURLLocation probe onto the same classifier with bounded backoff
and redacted URLs. Make the retry delays injectable so the tests stop
sleeping through the real backoff, and assert the hijack error on the
test goroutine.

Author: erezrokah

managedplugin/download.go

@@ -71,28 +71,31 @@ func getURLLocation(ctx context.Context, org string, name string, version string
 	var (
 		err404 = errors.New("404")
 		err401 = errors.New("401")
-		err429 = errors.New("429")
 	)
 
 	options := []retry.Option{
 		retry.RetryIf(func(err error) bool {
-			return err == err401 || err == err429
+			// The classifier treats 401 as permanent; this probe has always
+			// retried it because the GitHub asset host returns it spuriously.
+			return errors.Is(err, err401) || isRetryableDownloadError(err)
 		}),
 		retry.Context(ctx),
-		retry.Attempts(RetryAttempts),
-		retry.Delay(RetryWaitTime),
+		retry.Attempts(downloadRetryAttempts),
+		retry.Delay(downloadRetryDelay),
+		retry.MaxDelay(downloadRetryMaxDelay),
 		retry.LastErrorOnly(true),
 	}
 	retrier := retry.New(options...)
 	for _, downloadURL := range urls {
+		urlForLog := redactURLQuery(downloadURL)
 		err := retrier.Do(func() error {
 			req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
 			if err != nil {
-				return fmt.Errorf("failed create request %s: %w", downloadURL, err)
+				return fmt.Errorf("failed create request %s: %w", urlForLog, redactURLError(err))
 			}
 			resp, err := http.DefaultClient.Do(req)
 			if err != nil {
-				return fmt.Errorf("failed to get url %s: %w", downloadURL, err)
+				return fmt.Errorf("failed to get url %s: %w", urlForLog, redactURLError(err))
 			}
 			resp.Body.Close()
 			// Check server response
@@ -102,17 +105,18 @@ func getURLLocation(ctx context.Context, org string, name string, version string
 			case http.StatusNotFound:
 				return err404
 			case http.StatusUnauthorized:
-				fmt.Printf("Failed downloading %s with status code %d. Retrying\n", downloadURL, resp.StatusCode)
+				fmt.Printf("Failed downloading %s with status code %d. Retrying\n", urlForLog, resp.StatusCode)
 				return err401
-			case http.StatusTooManyRequests:
-				fmt.Printf("Failed downloading %s with status code %d. Retrying\n", downloadURL, resp.StatusCode)
-				return err429
 			default:
-				fmt.Printf("Failed downloading %s with status code %d\n", downloadURL, resp.StatusCode)
-				return fmt.Errorf("statusCode %d", resp.StatusCode)
+				if isRetryableStatusCode(resp.StatusCode) {
+					fmt.Printf("Failed downloading %s with status code %d. Retrying\n", urlForLog, resp.StatusCode)
+				} else {
+					fmt.Printf("Failed downloading %s with status code %d\n", urlForLog, resp.StatusCode)
+				}
+				return &httpStatusError{statusCode: resp.StatusCode}
 			}
 		})
-		if err == err404 {
+		if errors.Is(err, err404) {
 			continue
 		}
 		return downloadURL, err
@@ -346,9 +350,9 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop
 	options := []retry.Option{
 		retry.RetryIf(isRetryableDownloadError),
 		retry.Context(ctx),
-		retry.Attempts(RetryAttempts),
-		retry.Delay(RetryWaitTime),
-		retry.MaxDelay(MaxRetryWaitTime),
+		retry.Attempts(downloadRetryAttempts),
+		retry.Delay(downloadRetryDelay),
+		retry.MaxDelay(downloadRetryMaxDelay),
 	}
 	retrier := retry.New(options...)
 	err = retrier.Do(func() error {
@@ -362,13 +366,13 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop
 		// Get the data
 		req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
 		if err != nil {
-			return fmt.Errorf("failed create request %s: %w", urlForLog, err)
+			return fmt.Errorf("failed create request %s: %w", urlForLog, redactURLError(err))
 		}
 
 		// Do http request
 		resp, err := http.DefaultClient.Do(req)
 		if err != nil {
-			return fmt.Errorf("failed to get url %s: %w", urlForLog, err)
+			return fmt.Errorf("failed to get url %s: %w", urlForLog, redactURLError(err))
 		}
 		defer resp.Body.Close()
 		// Check server response

managedplugin/download_retry.go

@@ -17,6 +17,13 @@ var (
 	errShortRead = errors.New("truncated response body")
 )
 
+// Overridable so tests do not pay the real backoff.
+var (
+	downloadRetryAttempts = uint(RetryAttempts)
+	downloadRetryDelay    = RetryWaitTime
+	downloadRetryMaxDelay = MaxRetryWaitTime
+)
+
 type httpStatusError struct {
 	statusCode int
 }
@@ -34,8 +41,9 @@ func isRetryableStatusCode(statusCode int) bool {
 }
 
 // Go does not export the HTTP/2 stream and connection error types used by the
-// net/http transport, so the mid-body resets we get from the asset CDN can only
-// be matched on their message.
+// net/http transport, and on Windows the syscall.E* constants are synthetic
+// values that never match a real WSA socket error, so these failures can only be
+// matched on their message.
 var transientTransportMessages = []string{
 	"stream error",
 	"server sent goaway",
@@ -46,6 +54,14 @@ var transientTransportMessages = []string{
 	"server closed idle connection",
 	"transport connection broken",
 	"i/o timeout",
+	"connection refused",
+	"no such host",
+	"network is unreachable",
+	"no route to host",
+	// Windows WSAECONNRESET, WSAECONNREFUSED and WSAETIMEDOUT respectively.
+	"forcibly closed by the remote host",
+	"actively refused it",
+	"did not properly respond after a period of time",
 }
 
 func isRetryableDownloadError(err error) bool {
@@ -69,8 +85,18 @@ func isRetryableDownloadError(err error) bool {
 		errors.Is(err, io.ErrUnexpectedEOF),
 		errors.Is(err, io.EOF),
 		errors.Is(err, syscall.ECONNRESET),
+		errors.Is(err, syscall.ECONNREFUSED),
 		errors.Is(err, syscall.EPIPE),
-		errors.Is(err, syscall.ETIMEDOUT):
+		errors.Is(err, syscall.ETIMEDOUT),
+		errors.Is(err, syscall.EHOSTUNREACH),
+		errors.Is(err, syscall.ENETUNREACH):
+		return true
+	}
+
+	// The asset host is fixed, so a resolution failure against it is a resolver
+	// problem rather than a bad name.
+	var dnsErr *net.DNSError
+	if errors.As(err, &dnsErr) {
 		return true
 	}
 
@@ -100,6 +126,17 @@ func redactURLQuery(rawURL string) string {
 	return parsed.String()
 }
 
+// redactURLError rewrites the URL that *url.Error prints verbatim. Wrapping such
+// an error re-exposes the signed token that redactURLQuery removed from the
+// surrounding message.
+func redactURLError(err error) error {
+	var urlErr *url.Error
+	if errors.As(err, &urlErr) {
+		urlErr.URL = redactURLQuery(urlErr.URL)
+	}
+	return err
+}
+
 // IsTransientDownloadError reports whether err is a transient plugin download
 // failure - a network or server-side problem rather than a bad plugin reference.
 // Callers use it to keep advice about plugin resolution off errors that have

managedplugin/download_retry_test.go

@@ -11,8 +11,10 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"syscall"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -80,9 +82,14 @@ func TestRedactURLQuery(t *testing.T) {
 // attempt writes part of the body and then the connection drops mid-copy. The retry
 // must start the file from scratch rather than append to the partial bytes.
 func TestDownloadFileRetriesTruncatedBody(t *testing.T) {
+	fastRetries(t)
+
 	body := []byte("cloudquery-plugin-binary-payload")
 
-	var attempts int
+	var (
+		attempts  int
+		hijackErr error
+	)
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		attempts++
 		if attempts == 1 {
@@ -92,7 +99,10 @@ func TestDownloadFileRetriesTruncatedBody(t *testing.T) {
 			w.(http.Flusher).Flush()
 			// Close the connection mid-body so the client sees a truncated response.
 			conn, _, err := w.(http.Hijacker).Hijack()
-			require.NoError(t, err)
+			if err != nil {
+				hijackErr = err
+				return
+			}
 			conn.Close()
 			return
 		}
@@ -102,6 +112,7 @@ func TestDownloadFileRetriesTruncatedBody(t *testing.T) {
 
 	localPath := filepath.Join(t.TempDir(), "plugin.zip")
 	checksum, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true})
+	require.NoError(t, hijackErr)
 	require.NoError(t, err)
 	require.Equal(t, 2, attempts)
 
@@ -141,6 +152,8 @@ func TestDownloadFileDoesNotRetryNotFound(t *testing.T) {
 }
 
 func TestDownloadFileRetriesServerError(t *testing.T) {
+	fastRetries(t)
+
 	body := []byte("payload")
 
 	var attempts int
@@ -162,6 +175,8 @@ func TestDownloadFileRetriesServerError(t *testing.T) {
 }
 
 func TestDownloadFileGivesUpAfterRetryAttempts(t *testing.T) {
+	fastRetries(t)
+
 	var attempts int
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		attempts++
@@ -182,3 +197,97 @@ func sha256Hex(b []byte) string {
 	s.Write(b)
 	return fmt.Sprintf("%x", s.Sum(nil))
 }
+
+// fastRetries removes the real backoff so the retry tests do not sleep through it.
+func fastRetries(t *testing.T) {
+	t.Helper()
+
+	delay, maxDelay := downloadRetryDelay, downloadRetryMaxDelay
+	downloadRetryDelay, downloadRetryMaxDelay = time.Millisecond, time.Millisecond
+	t.Cleanup(func() {
+		downloadRetryDelay, downloadRetryMaxDelay = delay, maxDelay
+	})
+}
+
+// TestDownloadFileRedactsSignedTokenFromTransportErrors covers the leak that
+// url.Error reopens: it prints its URL verbatim, so wrapping one puts the signed
+// token back into the message once per attempt.
+func TestDownloadFileRedactsSignedTokenFromTransportErrors(t *testing.T) {
+	fastRetries(t)
+
+	const token = "SUPERSECRETTOKEN"
+
+	server := httptest.NewUnstartedServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {}))
+	server.Config.ConnState = func(c net.Conn, state http.ConnState) {
+		if state == http.StateActive {
+			c.Close()
+		}
+	}
+	server.Start()
+	t.Cleanup(server.Close)
+
+	localPath := filepath.Join(t.TempDir(), "plugin.zip")
+	_, err := downloadFile(context.Background(), localPath, server.URL+"/asset?verify="+token, DownloaderOptions{NoProgress: true})
+	require.Error(t, err)
+	require.NotContains(t, err.Error(), token)
+	require.NotContains(t, err.Error(), "verify=")
+}
+
+func TestDownloadFileRetriesConnectionRefused(t *testing.T) {
+	fastRetries(t)
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	addr := listener.Addr().String()
+	require.NoError(t, listener.Close())
+
+	localPath := filepath.Join(t.TempDir(), "plugin.zip")
+	_, err = downloadFile(context.Background(), localPath, "http://"+addr+"/asset", DownloaderOptions{NoProgress: true})
+	require.Error(t, err)
+	require.Equal(t, int(downloadRetryAttempts), strings.Count(err.Error(), "connection refused"))
+}
+
+func TestIsRetryableDownloadErrorWindowsSocketMessages(t *testing.T) {
+	cases := []struct {
+		name string
+		msg  string
+	}{
+		{name: "WSAECONNRESET", msg: "wsarecv: An existing connection was forcibly closed by the remote host."},
+		{name: "WSAECONNREFUSED", msg: "connectex: No connection could be made because the target machine actively refused it."},
+		{name: "WSAETIMEDOUT", msg: "connectex: A connection attempt failed because the connected party did not properly respond after a period of time."},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			require.True(t, isRetryableDownloadError(fmt.Errorf("failed to get url: %w", errors.New(tc.msg))))
+		})
+	}
+}
+
+func TestIsRetryableDownloadErrorDNS(t *testing.T) {
+	require.True(t, isRetryableDownloadError(fmt.Errorf("dial: %w", &net.DNSError{Err: "no such host", Name: "assets.cloudquery.io", IsNotFound: true})))
+}
+
+// TestTransportErrorFromRealRequestIsRetryable pins the gap that left
+// getURLLocation aborting on the first attempt: a transport failure surfaces as a
+// wrapped *url.Error, which its old identity comparison could never match.
+func TestTransportErrorFromRealRequestIsRetryable(t *testing.T) {
+	server := httptest.NewUnstartedServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {}))
+	server.Config.ConnState = func(c net.Conn, state http.ConnState) {
+		if state == http.StateActive {
+			c.Close()
+		}
+	}
+	server.Start()
+	t.Cleanup(server.Close)
+
+	resp, err := http.Get(server.URL + "/asset?verify=SUPERSECRETTOKEN")
+	if resp != nil {
+		resp.Body.Close()
+	}
+	require.Error(t, err)
+
+	wrapped := fmt.Errorf("failed to get url %s: %w", redactURLQuery(server.URL), redactURLError(err))
+	require.True(t, isRetryableDownloadError(wrapped))
+	require.NotContains(t, wrapped.Error(), "SUPERSECRETTOKEN")
+}