fetch: client timeout, retry with backoff, url.JoinPath (closes #63)
check / check (push) Failing after 2s
check / check (push) Failing after 2s
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
This commit is contained in:
+172
-56
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user