fetch: client timeout, retry with backoff, url.JoinPath (closes #63)

fetch now makes every request through an http.Client with a time
limit, ten minutes by default and set with --timeout, which must be
greater than zero. A connection error, a timeout, or a 5xx or 429
response is retried, up to five tries in all, after a random wait
whose limit doubles from one second, or after the wait the server's
Retry-After asks for, up to one minute. Each try of a file starts a
new temp file, so a retry never keeps a partial file; the size and
hash checks are unchanged. Manifest and file URLs are built with
URL.JoinPath, so trailing slashes, query strings and names that need
escaping all work.

Model: opus-5-5

internal/cli/fetch.go

@@ -7,11 +7,13 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/rand/v2"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
-	"path"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -45,12 +47,33 @@ const (
 	bpsPerGbps = 1e9
 	bpsPerMbps = 1e6
 	bpsPerKbps = 1e3
+
+	// httpTimeout is the default time limit for one HTTP request, from
+	// connecting to reading the last byte of the body. It also bounds how
+	// long one file may take to download; fetch's --timeout changes it.
+	httpTimeout = 10 * time.Minute
+
+	// fetchAttempts is how many times fetch tries a request before it
+	// gives up on a transient failure.
+	fetchAttempts = 5
+
+	// firstRetryDelay is the longest fetch waits before its first retry.
+	// The limit doubles for each retry after that; the wait itself is
+	// random up to the limit.
+	firstRetryDelay = time.Second
+
+	// maxRetryAfter is the longest wait a server's Retry-After header can
+	// ask for. A server asking for longer fails the request at once.
+	maxRetryAfter = time.Minute
 )
 
 var (
 	// errURLRequired indicates the fetch command was run without a URL
 	// argument.
 	errURLRequired = errors.New("URL argument required")
+	// errInvalidTimeout indicates a fetch --timeout of zero or less, which
+	// http.Client would take as no time limit at all.
+	errInvalidTimeout = errors.New("--timeout must be greater than zero")
 	// errEmptyPath indicates an empty file path in the manifest.
 	errEmptyPath = errors.New("empty path")
 	// errAbsolutePath indicates an absolute file path in the manifest.
@@ -78,24 +101,109 @@ type DownloadProgress struct {
 	ETA         time.Duration // Estimated time to completion
 }
 
-// httpGet issues a GET request for the given URL using the provided
-// context and returns the response. The caller must close the body.
+// retryingClient is the HTTP client fetch makes every request with. It
+// waits up to firstDelay, which must be positive, before its first retry
+// of a transient failure. Tests shorten both the client's timeout and
+// firstDelay.
+type retryingClient struct {
+	client     *http.Client
+	firstDelay time.Duration
+}
+
+// get issues a GET for rawURL and passes a 200 OK response to use.
 //
-// Errors are returned unwrapped: this helper replaced direct http.Get
-// calls, and each caller already supplies its own context string, so
-// adding one here would change user-visible messages.
-func httpGet(ctx context.Context, fileURL string) (*http.Response, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fileURL, nil)
+// A connection error, a timeout, or a 5xx or 429 status is retried, up to
+// fetchAttempts tries in all. Before each retry it waits a random time up
+// to a limit that doubles each time, unless the server's Retry-After
+// header says how long to wait. Any other failure, an error from use
+// included, is returned at once and unwrapped; a non-OK status is
+// returned as errHTTPStatus. use must start over each time it is called,
+// since a retry calls it again with a new response.
+func (c retryingClient) get(
+	ctx context.Context, rawURL string, use func(*http.Response) error,
+) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return nil, err
+	delay := c.firstDelay
+
+	for attempt := 1; ; attempt++ {
+		// Wait a random time up to delay, so that clients that failed
+		// together do not all retry together.
+		wait := rand.N(delay) //nolint:gosec // G404: jitter, not a secret
+
+		var retry bool
+
+		resp, err := c.client.Do(req)
+
+		switch {
+		case err != nil:
+			retry = isConnectionError(err)
+		case resp.StatusCode == http.StatusOK:
+			err = use(resp)
+			retry = isConnectionError(err)
+
+			_ = resp.Body.Close()
+		default:
+			_ = resp.Body.Close()
+
+			err = fmt.Errorf("%w %d", errHTTPStatus, resp.StatusCode)
+			retry = resp.StatusCode >= http.StatusInternalServerError ||
+				resp.StatusCode == http.StatusTooManyRequests
+
+			after, ok := retryAfter(resp.Header.Get("Retry-After"))
+			if ok {
+				wait = after
+				retry = retry && after <= maxRetryAfter
+			}
+		}
+
+		if !retry || attempt == fetchAttempts || ctx.Err() != nil {
+			return err
+		}
+
+		log.Warnf("%s: %s, retrying in %s", rawURL, err, wait.Round(time.Millisecond))
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(wait):
+		}
+
+		delay *= 2
+	}
+}
+
+// isConnectionError reports whether err is a connection error or a
+// timeout: a connection that could not be made, was reset, or closed
+// early, or a request that ran out of time.
+func isConnectionError(err error) bool {
+	var (
+		opErr  *net.OpError
+		netErr net.Error
+	)
+
+	return errors.As(err, &opErr) ||
+		(errors.As(err, &netErr) && netErr.Timeout()) ||
+		errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)
+}
+
+// retryAfter returns the wait a Retry-After header value asks for, given
+// either in seconds or as an HTTP date. ok is false if there is none.
+func retryAfter(value string) (time.Duration, bool) {
+	seconds, err := strconv.Atoi(value)
+	if err == nil {
+		return time.Duration(seconds) * time.Second, true
 	}
 
-	return resp, nil
+	when, err := http.ParseTime(value)
+	if err == nil {
+		return time.Until(when), true
+	}
+
+	return 0, false
 }
 
 // reportDownloadProgress renders download progress until the channel
@@ -119,25 +227,23 @@ func reportDownloadProgress(progress <-chan DownloadProgress, done chan<- struct
 }
 
 // manifestBaseURL returns the URL of the directory containing the
-// manifest, with a trailing slash.
+// manifest.
 func manifestBaseURL(manifestURL string) (*url.URL, error) {
-	baseURL, err := url.Parse(manifestURL)
+	parsed, err := url.Parse(manifestURL)
 	if err != nil {
 		return nil, fmt.Errorf("fetch: invalid manifest URL: %w", err)
 	}
 
-	baseURL.Path = path.Dir(baseURL.Path)
-	if !strings.HasSuffix(baseURL.Path, "/") {
-		baseURL.Path += "/"
-	}
-
-	return baseURL, nil
+	// JoinPath cleans the path it builds, so ".." drops the manifest's
+	// file name.
+	return parsed.JoinPath(".."), nil
 }
 
 // downloadManifestFiles downloads every file in the manifest, reporting
 // progress on the progress channel.
 func downloadManifestFiles(
 	ctx context.Context,
+	client retryingClient,
 	baseURL *url.URL,
 	files []*mfer.MFFilePath,
 	progress chan<- DownloadProgress,
@@ -149,10 +255,12 @@ func downloadManifestFiles(
 			return fmt.Errorf("invalid path in manifest: %w", err)
 		}
 
-		fileURL := baseURL.String() + encodeFilePath(f.GetPath())
+		// JoinPath takes escaped path text, so a name such as "100%.txt"
+		// must be escaped first.
+		fileURL := baseURL.JoinPath(encodeFilePath(f.GetPath())).String()
 		log.Infof("fetching %s", f.GetPath())
 
-		err = downloadFile(ctx, fileURL, localPath, f, progress)
+		err = downloadFile(ctx, client, fileURL, localPath, f, progress)
 		if err != nil {
 			return fmt.Errorf("failed to download %s: %w", f.GetPath(), err)
 		}
@@ -168,30 +276,41 @@ func (mfa *CLIApp) fetchManifestOperation(ctx *cli.Context) error {
 		return errURLRequired
 	}
 
-	inputURL := ctx.Args().Get(0)
+	timeout := ctx.Duration(flagTimeout)
+	if timeout <= 0 {
+		return errInvalidTimeout
+	}
 
-	manifestURL, err := resolveManifestURL(inputURL)
+	manifestURL, err := resolveManifestURL(ctx.Args().Get(0))
 	if err != nil {
 		return fmt.Errorf("invalid URL: %w", err)
 	}
 
+	client := retryingClient{
+		client:     &http.Client{Timeout: timeout},
+		firstDelay: firstRetryDelay,
+	}
+
 	log.Infof("fetching manifest from %s", manifestURL)
 
-	// Fetch manifest
-	resp, err := httpGet(ctx.Context, manifestURL)
+	// Read the whole manifest before parsing it, so that a connection
+	// lost partway through is retried rather than reported as a bad
+	// manifest.
+	var manifestData []byte
+
+	err = client.get(ctx.Context, manifestURL, func(resp *http.Response) error {
+		var readErr error
+
+		manifestData, readErr = io.ReadAll(resp.Body)
+
+		return readErr
+	})
 	if err != nil {
 		return fmt.Errorf("failed to fetch manifest: %w", err)
 	}
 
-	defer func() { _ = resp.Body.Close() }()
-
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("failed to fetch manifest: %w %d",
-			errHTTPStatus, resp.StatusCode)
-	}
-
 	// Parse manifest
-	manifest, err := mfer.NewManifestFromReader(resp.Body)
+	manifest, err := mfer.NewManifestFromReader(bytes.NewReader(manifestData))
 	if err != nil {
 		return fmt.Errorf("failed to parse manifest: %w", err)
 	}
@@ -221,7 +340,7 @@ func (mfa *CLIApp) fetchManifestOperation(ctx *cli.Context) error {
 	startTime := time.Now()
 
 	// Download each file
-	dlErr := downloadManifestFiles(ctx.Context, baseURL, files, progress)
+	dlErr := downloadManifestFiles(ctx.Context, client, baseURL, files, progress)
 
 	close(progress)
 	<-done
@@ -325,14 +444,7 @@ func resolveManifestURL(inputURL string) (string, error) {
 		return inputURL, nil
 	}
 
-	// Ensure path ends with /
-	if !strings.HasSuffix(parsed.Path, "/") {
-		parsed.Path += "/"
-	}
-
-	parsed.Path += defaultManifestName
-
-	return parsed.String(), nil
+	return parsed.JoinPath(defaultManifestName).String(), nil
 }
 
 // progressWriter wraps an io.Writer and reports progress to a channel.
@@ -441,6 +553,7 @@ func verifyDownloadedHash(digest []byte, entry *mfer.MFFilePath) error {
 // Progress is reported via the progress channel.
 func downloadFile(
 	ctx context.Context,
+	client retryingClient,
 	fileURL, localPath string,
 	entry *mfer.MFFilePath,
 	progress chan<- DownloadProgress,
@@ -468,18 +581,21 @@ func downloadFile(
 
 	tmpPath := tempPathFor(localPath)
 
-	// Fetch file
-	resp, err := httpGet(ctx, fileURL)
-	if err != nil {
-		return fmt.Errorf("HTTP request failed: %w", err)
-	}
-
-	defer func() { _ = resp.Body.Close() }()
-
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("%w %d", errHTTPStatus, resp.StatusCode)
-	}
+	return client.get(ctx, fileURL, func(resp *http.Response) error {
+		return saveResponse(resp, tmpPath, localPath, entry, progress)
+	})
+}
 
+// saveResponse writes resp's body to tmpPath, verifies it against entry,
+// and renames it to localPath. It starts a new temp file each time and
+// removes it on failure, so a retry after a failed try never appends to
+// or keeps a partial file.
+func saveResponse(
+	resp *http.Response,
+	tmpPath, localPath string,
+	entry *mfer.MFFilePath,
+	progress chan<- DownloadProgress,
+) error {
 	// Determine expected size
 	expectedSize := entry.GetSize()
 
@@ -488,7 +604,7 @@ func downloadFile(
 		totalBytes = expectedSize
 	}
 
-	err = checkNoSymlinks(tmpPath)
+	err := checkNoSymlinks(tmpPath)
 	if err != nil {
 		return err
 	}