source/http: honor Retry-After for rate limits and outages

Retry HTTP 429 and 503 responses through the shared retry loop, honoring
Retry-After seconds and HTTP dates without extending the attempt limit.
Allow server-requested delays up to 30 seconds independently of the
exponential backoff limit. Stop when the requested delay exceeds this
limit instead of retrying earlier than the server requested.

Close failed response bodies before waiting and preserve cancellation.
Keep the cached HEAD probe best-effort, with retries applied to GET.

Signed-off-by: CrazyMax <1951866+crazy-max@users.noreply.github.com>

source/http/retry_test.go

@@ -1,11 +1,18 @@
 package http
 
 import (
+	"context"
+	"io"
+	"math"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
+	"time"
 
+	"github.com/moby/buildkit/util/resolver/retryhandler"
 	"github.com/stretchr/testify/require"
 )
 
@@ -46,9 +53,7 @@ func TestDoRequestRetriesDroppedConnection(t *testing.T) {
 	require.Equal(t, int32(3), attempts.Load())
 }
 
-// TestDoRequestDoesNotRetryStatusErrors asserts the retry does not change how
-// HTTP status codes are handled: the response is returned to the caller on the
-// first attempt, exactly as before.
+// TestDoRequestDoesNotRetryStatusErrors checks a permanent HTTP failure.
 func TestDoRequestDoesNotRetryStatusErrors(t *testing.T) {
 	t.Parallel()
 
@@ -70,3 +75,222 @@ func TestDoRequestDoesNotRetryStatusErrors(t *testing.T) {
 	require.Equal(t, http.StatusNotFound, resp.StatusCode)
 	require.Equal(t, int32(1), attempts.Load())
 }
+
+func TestDoRequestRetryAfter(t *testing.T) {
+	for _, method := range []string{http.MethodGet, http.MethodHead} {
+		for _, status := range []int{http.StatusTooManyRequests, http.StatusServiceUnavailable} {
+			t.Run(method+"/"+http.StatusText(status), func(t *testing.T) {
+				synctest.Test(t, func(t *testing.T) {
+					attempts := 0
+					start := time.Now()
+					failedBody := &retryTestBody{Reader: strings.NewReader("retry later")}
+					successBody := &retryTestBody{Reader: strings.NewReader("content")}
+					client := &http.Client{Transport: retryTestTransport(func(req *http.Request) (*http.Response, error) {
+						attempts++
+						require.Equal(t, method, req.Method)
+						require.Equal(t, "etag", req.Header.Get("If-None-Match"))
+						if attempts == 1 {
+							return &http.Response{
+								StatusCode: status,
+								Header:     http.Header{"Retry-After": {"10"}},
+								Body:       failedBody,
+							}, nil
+						}
+						require.True(t, failedBody.closed)
+						require.Equal(t, len("retry later"), failedBody.Len())
+						require.Equal(t, 10*time.Second, time.Since(start))
+						return &http.Response{StatusCode: http.StatusOK, Body: successBody}, nil
+					})}
+					req, err := http.NewRequestWithContext(t.Context(), method, "https://example.com", nil)
+					require.NoError(t, err)
+					req.Header.Set("If-None-Match", "etag")
+					resp, err := doRequest(t.Context(), client, req)
+					require.NoError(t, err)
+					defer resp.Body.Close()
+					require.Equal(t, 2, attempts)
+					require.False(t, successBody.closed)
+					body, err := io.ReadAll(resp.Body)
+					require.NoError(t, err)
+					require.Equal(t, "content", string(body))
+				})
+			})
+		}
+	}
+}
+
+func TestDoRequestStatusRetryLimit(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		status     int
+		retryAfter string
+		attempts   int
+		wantError  bool
+	}{
+		{
+			name:      "rate limited",
+			status:    http.StatusTooManyRequests,
+			attempts:  4,
+			wantError: true,
+		},
+		{
+			name:       "unavailable with delay",
+			status:     http.StatusServiceUnavailable,
+			retryAfter: "3",
+			attempts:   4,
+			wantError:  true,
+		},
+		{
+			name:       "delay exceeds limit",
+			status:     http.StatusTooManyRequests,
+			retryAfter: "3600",
+			attempts:   1,
+			wantError:  true,
+		},
+		{
+			name:       "other status ignores retry after",
+			status:     http.StatusInternalServerError,
+			retryAfter: "3",
+			attempts:   1,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				attempts := 0
+				start := time.Now()
+				var bodies []*retryTestBody
+				client := &http.Client{Transport: retryTestTransport(func(*http.Request) (*http.Response, error) {
+					attempts++
+					body := &retryTestBody{Reader: strings.NewReader("error")}
+					bodies = append(bodies, body)
+					return &http.Response{
+						StatusCode: tc.status,
+						Header:     http.Header{"Retry-After": {tc.retryAfter}},
+						Body:       body,
+					}, nil
+				})}
+				req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "https://example.com", nil)
+				require.NoError(t, err)
+				resp, err := doRequest(t.Context(), client, req)
+				if tc.wantError {
+					var retryAfter *retryhandler.RetryAfterError
+					require.ErrorAs(t, err, &retryAfter)
+					require.Nil(t, resp)
+					for _, body := range bodies {
+						require.True(t, body.closed)
+					}
+				} else {
+					require.NoError(t, err)
+					require.Equal(t, tc.status, resp.StatusCode)
+					require.False(t, bodies[0].closed)
+					resp.Body.Close()
+				}
+				require.Equal(t, tc.attempts, attempts)
+				if tc.attempts == 1 {
+					require.Zero(t, time.Since(start))
+				}
+			})
+		})
+	}
+}
+
+func TestDoRequestRetryAfterCancellation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ctx, cancel := context.WithCancelCause(t.Context())
+		defer cancel(context.Canceled)
+		attempts := 0
+		body := &retryTestBody{Reader: strings.NewReader("retry later")}
+		client := &http.Client{Transport: retryTestTransport(func(*http.Request) (*http.Response, error) {
+			attempts++
+			return &http.Response{
+				StatusCode: http.StatusServiceUnavailable,
+				Header:     http.Header{"Retry-After": {"30"}},
+				Body:       body,
+			}, nil
+		})}
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://example.com", nil)
+		require.NoError(t, err)
+		done := make(chan error, 1)
+		go func() {
+			resp, err := doRequest(ctx, client, req)
+			if resp != nil {
+				resp.Body.Close()
+			}
+			done <- err
+		}()
+		synctest.Wait()
+		require.True(t, body.closed)
+		start := time.Now()
+		cancel(context.Canceled)
+		require.ErrorIs(t, <-done, context.Canceled)
+		require.Zero(t, time.Since(start))
+		require.Equal(t, 1, attempts)
+	})
+}
+
+func TestRetryAfterDelay(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		for _, tc := range []struct {
+			name  string
+			value string
+			want  time.Duration
+		}{
+			{
+				name:  "seconds",
+				value: "3",
+				want:  3 * time.Second,
+			},
+			{
+				name:  "date",
+				value: time.Now().Add(5 * time.Second).UTC().Format(http.TimeFormat),
+				want:  5 * time.Second,
+			},
+			{
+				name:  "past date",
+				value: time.Now().Add(-time.Minute).UTC().Format(http.TimeFormat),
+			},
+			{
+				name: "missing",
+			},
+			{
+				name:  "zero",
+				value: "0",
+			},
+			{
+				name:  "invalid",
+				value: "tomorrow",
+			},
+			{
+				name:  "negative",
+				value: "-1",
+			},
+			{
+				name:  "duration overflow",
+				value: "9223372037",
+				want:  time.Duration(math.MaxInt64),
+			},
+			{
+				name:  "integer overflow",
+				value: "18446744073709551616",
+				want:  time.Duration(math.MaxInt64),
+			},
+		} {
+			require.Equal(t, tc.want, retryAfterDelay(tc.value), tc.name)
+		}
+	})
+}
+
+type retryTestTransport func(*http.Request) (*http.Response, error)
+
+func (f retryTestTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+type retryTestBody struct {
+	*strings.Reader
+	closed bool
+}
+
+func (b *retryTestBody) Close() error {
+	b.closed = true
+	return nil
+}

source/http/source.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"math"
 	"mime"
 	"net/http"
 	"net/url"
@@ -216,24 +217,45 @@ func (hs *httpSourceHandler) client(g session.Group) *http.Client {
 	return &http.Client{Transport: newTransport(hs.transport, hs.sm, g)}
 }
 
-// doRequest performs req, retrying transient network failures with
-// exponential backoff.
-//
-// A solve can contain thousands of independent HTTP sources, each fetched over
-// its own connection and with no connection reuse between them. Without a
-// retry, one dropped connection anywhere fails the entire solve. Registry
-// fetches already retry this class of error through retryhandler; HTTP sources
-// did not.
-//
-// Only the request is retried. A failure part way through reading the response
-// body is not covered here, and neither is a non-2xx response: those are left
-// to the caller so that status handling is unchanged.
+// doRequest retries transient network failures and HTTP 429/503 responses with
+// jittered exponential backoff, honoring Retry-After as a minimum delay.
+// Requests have no body and are safe to replay. Failures reading a successful
+// response body and other HTTP statuses are left to the caller.
 func doRequest(ctx context.Context, client *http.Client, req *http.Request) (*http.Response, error) {
 	return retryhandler.WithRetry(ctx, nil, func(ctx context.Context) (*http.Response, error) {
-		return client.Do(req.Clone(ctx))
+		resp, err := client.Do(req.Clone(ctx))
+		if err != nil {
+			return nil, err
+		}
+		if resp.StatusCode != http.StatusTooManyRequests && resp.StatusCode != http.StatusServiceUnavailable {
+			return resp, nil
+		}
+		// Do not drain an error body: it may stall independently of the headers.
+		resp.Body.Close()
+		return nil, &retryhandler.RetryAfterError{
+			Err:   errors.Errorf("invalid response status %d", resp.StatusCode),
+			Delay: retryAfterDelay(resp.Header.Get("Retry-After")),
+		}
 	})
 }
 
+// retryAfterDelay parses either delay-seconds or an HTTP-date. Invalid and past
+// values fall back to the normal backoff. Saturate large values to avoid overflow.
+func retryAfterDelay(value string) time.Duration {
+	value = strings.TrimSpace(value)
+	seconds, err := strconv.ParseUint(value, 10, 64)
+	if err == nil || errors.Is(err, strconv.ErrRange) {
+		if seconds > uint64(math.MaxInt64/int64(time.Second)) {
+			return time.Duration(math.MaxInt64)
+		}
+		return time.Duration(seconds) * time.Second
+	}
+	if date, err := http.ParseTime(value); err == nil {
+		return max(time.Until(date), 0)
+	}
+	return 0
+}
+
 // urlHash is internal hash the etag is stored by that doesn't leak outside
 // this package.
 func (hs *httpSourceHandler) urlHash() (digest.Digest, error) {

source/http/source_test.go

@@ -2,8 +2,11 @@ package http
 
 import (
 	"context"
+	"io"
+	"net/http"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/containerd/containerd/v2/core/diff/apply"
@@ -31,6 +34,49 @@ import (
 
 const signFixturesPathEnv = "BUILDKIT_TEST_SIGN_FIXTURES"
 
+func TestHTTPSourceBestEffortHEAD(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+	cm, err := newCacheManager(t)
+	require.NoError(t, err)
+	attempts := 0
+	hs, err := NewSource(Opt{
+		CacheAccessor: cm,
+		Transport: retryTestTransport(func(req *http.Request) (*http.Response, error) {
+			attempts++
+			if attempts != 2 {
+				require.Equal(t, http.MethodGet, req.Method)
+				return &http.Response{
+					StatusCode:    http.StatusOK,
+					Header:        http.Header{"Etag": {`"etag"`}},
+					Body:          io.NopCloser(strings.NewReader("content")),
+					ContentLength: int64(len("content")),
+				}, nil
+			}
+			require.Equal(t, http.MethodHead, req.Method)
+			return &http.Response{
+				StatusCode: http.StatusTooManyRequests,
+				Header:     http.Header{"Retry-After": {"3600"}},
+				Body:       http.NoBody,
+			}, nil
+		}),
+	})
+	require.NoError(t, err)
+	id := &HTTPIdentifier{URL: "https://example.com/file"}
+	h, err := hs.Resolve(ctx, id, nil, nil)
+	require.NoError(t, err)
+	_, _, _, _, err = h.CacheKey(ctx, nil, 0)
+	require.NoError(t, err)
+
+	// Resolve again to probe the cached ETag with HEAD. The probe is best-effort,
+	// so a rate-limited response falls through to GET without retrying HEAD.
+	h, err = hs.Resolve(ctx, id, nil, nil)
+	require.NoError(t, err)
+	_, _, _, _, err = h.CacheKey(ctx, nil, 0)
+	require.NoError(t, err)
+	require.Equal(t, 3, attempts)
+}
+
 func TestHTTPSource(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()

util/resolver/retryhandler/retry.go

@@ -20,6 +20,27 @@ import (
 // variable so that code which embeds BuildKit can override the default value.
 var MaxRetryBackoff = 8 * time.Second
 
+// MaxRetryAfter is the longest server-requested delay accepted for a retry.
+// It is independent of the exponential backoff budget. Code which embeds
+// BuildKit can override the default value.
+var MaxRetryAfter = 30 * time.Second
+
+// RetryAfterError marks a failure as retryable after at least Delay.
+// A zero Delay uses the normal backoff. Delays above MaxRetryAfter
+// stop retries rather than retrying before the requested delay has elapsed.
+type RetryAfterError struct {
+	Err   error
+	Delay time.Duration
+}
+
+func (e *RetryAfterError) Error() string {
+	return e.Err.Error()
+}
+
+func (e *RetryAfterError) Unwrap() error {
+	return e.Err
+}
+
 func New(f images.HandlerFunc, logger func([]byte)) images.HandlerFunc {
 	return func(ctx context.Context, desc ocispecs.Descriptor) ([]ocispecs.Descriptor, error) {
 		descs, err := WithRetry(ctx, logger, func(ctx context.Context) ([]ocispecs.Descriptor, error) {
@@ -69,6 +90,12 @@ func WithRetryIf[T any](ctx context.Context, logger func([]byte), f func(context
 		}
 		// Keep the retry budget independent of jitter, including the first retry.
 		delay := backoff/2 + rand.N(backoff/2) //nolint:gosec // Jitter spreads load; it does not need cryptographic randomness.
+		if retryAfter, ok := stderrors.AsType[*RetryAfterError](err); ok {
+			if retryAfter.Delay > MaxRetryAfter {
+				return v, err
+			}
+			delay = max(delay, retryAfter.Delay)
+		}
 		if logger != nil {
 			logger(fmt.Appendf(nil, "retrying in %v\n", delay))
 		}
@@ -91,6 +118,10 @@ func contextFailure(ctx context.Context) error {
 }
 
 func retryError(err error) bool {
+	if _, ok := stderrors.AsType[*RetryAfterError](err); ok {
+		return true
+	}
+
 	// Retry on 5xx errors
 	var errUnexpectedStatus remoteserrors.ErrUnexpectedStatus
 	if errors.As(err, &errUnexpectedStatus) &&

util/resolver/retryhandler/retry_test.go

@@ -116,3 +116,73 @@ func TestWithRetryCancellationDuringBackoff(t *testing.T) {
 		require.Zero(t, time.Since(start))
 	})
 }
+
+func TestWithRetryAfter(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		delay    time.Duration
+		veto     bool
+		attempts int
+		elapsed  time.Duration
+	}{
+		{
+			name:     "server minimum",
+			delay:    7 * time.Second,
+			attempts: 4,
+			elapsed:  21 * time.Second,
+		},
+		{
+			name:     "at exponential backoff limit",
+			delay:    MaxRetryBackoff,
+			attempts: 4,
+			elapsed:  3 * MaxRetryBackoff,
+		},
+		{
+			name:     "above exponential backoff limit",
+			delay:    10 * time.Second,
+			attempts: 4,
+			elapsed:  30 * time.Second,
+		},
+		{
+			name:     "at server delay limit",
+			delay:    MaxRetryAfter,
+			attempts: 4,
+			elapsed:  3 * MaxRetryAfter,
+		},
+		{
+			name:     "just above server delay limit",
+			delay:    MaxRetryAfter + time.Nanosecond,
+			attempts: 1,
+		},
+		{
+			name:     "above limit",
+			delay:    time.Hour,
+			attempts: 1,
+		},
+		{
+			name:     "predicate veto",
+			delay:    time.Second,
+			veto:     true,
+			attempts: 1,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				cause := errors.New("retry later")
+				failure := errors.Wrap(&RetryAfterError{Err: cause, Delay: tc.delay}, "request failed")
+				attempts := 0
+				start := time.Now()
+				_, err := WithRetryIf(t.Context(), nil, func(ctx context.Context) (int, error) {
+					return WithRetry(ctx, nil, func(context.Context) (int, error) {
+						attempts++
+						return 0, failure
+					})
+				}, func(error) bool { return !tc.veto })
+				require.ErrorIs(t, err, cause)
+				require.Equal(t, failure.Error(), err.Error())
+				require.Equal(t, tc.attempts, attempts)
+				require.Equal(t, tc.elapsed, time.Since(start))
+			})
+		})
+	}
+}