429 Too Many Requests

package reposync import ( "context" "errors" "fmt" "io" "net" "net/http" "net/url" "strconv" "syscall" "testing" "time" "github.com/bluesky-social/indigo/xrpc" "github.com/ipfs/go-cid" "github.com/stretchr/testify/require" ) // fastRetry is the policy the network tests use: same shape as production, // milliseconds instead of seconds. func fastRetry() RetryPolicy { return RetryPolicy{MaxAttempts: 5, BaseDelay: time.Millisecond, MaxDelay: 20 * time.Millisecond} } // throttled builds the failure a rate-limiting host sends, optionally with the // ratelimit-* headers indigo knows how to parse. func throttled(reset time.Time) failure { f := failure{ status: http.StatusTooManyRequests, body: `{"error":"RateLimitExceeded","message":"Rate Limit Exceeded"}`, header: map[string]string{"Content-Type": "application/json"}, } if !reset.IsZero() { f.header["ratelimit-limit"] = "3000" f.header["ratelimit-remaining"] = "0" f.header["ratelimit-policy"] = "3000;w=300" f.header["ratelimit-reset"] = strconv.FormatInt(reset.Unix(), 10) } return f } // htmlThrottled is the shape that actually broke a production walk: a 429 whose // body is an HTML error page, so indigo cannot decode an XRPCError out of it and // only the status code survives. var htmlThrottled = failure{ status: http.StatusTooManyRequests, body: "429 Too Many Requestsgo away", header: map[string]string{"Content-Type": "text/html"}, } func xrpcErr(status int, errStr, msg string) error { return fmt.Errorf("getBlocks: %w", &xrpc.Error{ StatusCode: status, Wrapped: &xrpc.XRPCError{ErrStr: errStr, Message: msg}, }) } func TestIsRetryable(t *testing.T) { for _, tc := range []struct { name string err error want bool }{ {"nil", nil, false}, {"429", xrpcErr(http.StatusTooManyRequests, "RateLimitExceeded", "slow down"), true}, {"429 with an undecodable body", fmt.Errorf("getBlocks: %w", &xrpc.Error{ StatusCode: http.StatusTooManyRequests, Wrapped: errors.New("failed to decode xrpc error message: invalid character '<'"), }), true}, {"500", xrpcErr(http.StatusInternalServerError, "InternalServerError", "oops"), true}, {"502", fmt.Errorf("x: %w", &xrpc.Error{StatusCode: http.StatusBadGateway}), true}, {"503", fmt.Errorf("x: %w", &xrpc.Error{StatusCode: http.StatusServiceUnavailable}), true}, {"504", fmt.Errorf("x: %w", &xrpc.Error{StatusCode: http.StatusGatewayTimeout}), true}, // Numerically 5xx, but a permanent answer: the backfill wants to hear // it at once so it can fall back to getRepo. {"501", fmt.Errorf("x: %w", &xrpc.Error{StatusCode: http.StatusNotImplemented}), false}, {"400 could not find cids", xrpcErr(http.StatusBadRequest, "InvalidRequest", "Could not find cids: bafy"), false}, {"401", fmt.Errorf("x: %w", &xrpc.Error{StatusCode: http.StatusUnauthorized}), false}, {"404", fmt.Errorf("x: %w", &xrpc.Error{StatusCode: http.StatusNotFound}), false}, {"connection reset", fmt.Errorf("request failed: %w", syscall.ECONNRESET), true}, {"connection refused", fmt.Errorf("request failed: %w", syscall.ECONNREFUSED), true}, {"truncated body", fmt.Errorf("reading response body: %w", io.ErrUnexpectedEOF), true}, {"timeout", fmt.Errorf("request failed: %w", timeoutError{}), true}, {"canceled", fmt.Errorf("request failed: %w", context.Canceled), false}, {"deadline exceeded", fmt.Errorf("request failed: %w", context.DeadlineExceeded), false}, {"missing block", fmt.Errorf("x: %w", ErrMissingBlock), false}, {"block mismatch", fmt.Errorf("x: %w", ErrBlockMismatch), false}, {"plain error", errors.New("nope"), false}, } { t.Run(tc.name, func(t *testing.T) { require.Equal(t, tc.want, isRetryable(tc.err)) }) } } type timeoutError struct{} func (timeoutError) Error() string { return "i/o timeout" } func (timeoutError) Timeout() bool { return true } func (timeoutError) Temporary() bool { return true } func TestRetryDelay(t *testing.T) { p := RetryPolicy{BaseDelay: time.Second, MaxDelay: 30 * time.Second} plain := errors.New("boom") // Exponential, jittered down by at most 25%. for _, tc := range []struct{ attempt, wantSec int }{{1, 1}, {2, 2}, {3, 4}, {4, 8}, {5, 16}} { full := time.Duration(tc.wantSec) * time.Second for i := 0; i < 50; i++ { d, source := p.delay(tc.attempt, plain) require.GreaterOrEqual(t, d, time.Duration(float64(full)*0.75), "attempt %d", tc.attempt) require.LessOrEqual(t, d, full, "attempt %d", tc.attempt) require.Empty(t, source, "nothing told us to wait, so nothing is reported") } } // Capped, and still jittered at the cap so a fleet does not resynchronize. var sawJitter bool for i := 0; i < 50; i++ { d, _ := p.delay(10, plain) require.GreaterOrEqual(t, d, 22500*time.Millisecond) require.LessOrEqual(t, d, 30*time.Second) if d < 29*time.Second { sawJitter = true } } require.True(t, sawJitter) // The zero policy is the documented defaults. zero, _ := RetryPolicy{}.delay(1, plain) require.LessOrEqual(t, zero, DefaultRetryBaseDelay) require.GreaterOrEqual(t, zero, DefaultRetryBaseDelay*3/4) t.Run("ratelimit reset is honored", func(t *testing.T) { // Further out than the backoff for attempt 1 (~1s): wait for the reset. err := ratelimited(time.Now().Add(3 * time.Second)) d, source := p.delay(1, err) require.Greater(t, d, 2500*time.Millisecond) require.LessOrEqual(t, d, 3500*time.Millisecond) require.Equal(t, "ratelimit-reset", source) }) t.Run("ratelimit reset is clamped to MaxDelay", func(t *testing.T) { // bsky rate limit windows are minutes long; we would rather make one // more doomed attempt than hold a per-PDS lock that long. err := ratelimited(time.Now().Add(10 * time.Minute)) d, source := p.delay(1, err) require.Equal(t, 30*time.Second, d) require.Equal(t, "ratelimit-reset", source) }) t.Run("a reset in the past does not shorten the backoff", func(t *testing.T) { err := ratelimited(time.Now().Add(-time.Minute)) d, source := p.delay(3, err) require.GreaterOrEqual(t, d, 3*time.Second) require.LessOrEqual(t, d, 4*time.Second) require.Empty(t, source) }) t.Run("a hint from the registry is honored", func(t *testing.T) { hints := NewBackoffHints() hints.Observe("https://pds.example/", http.StatusTooManyRequests, http.Header{"Retry-After": []string{"3"}}) hinted := RetryPolicy{BaseDelay: time.Second, MaxDelay: 30 * time.Second, Hints: hints, Host: "https://pds.example"} d, source := hinted.delay(1, errors.New("boom")) require.Greater(t, d, 2500*time.Millisecond) require.LessOrEqual(t, d, 3500*time.Millisecond) require.Equal(t, "retry-after", source) // A hint that is shorter than the ladder changes nothing: the ladder is // the floor, the hint only ever pushes a wait out. d, source = hinted.delay(5, errors.New("boom")) require.GreaterOrEqual(t, d, 12*time.Second) require.Empty(t, source) // A different host is a different budget. other := hinted other.Host = "other.example" d, source = other.delay(1, errors.New("boom")) require.LessOrEqual(t, d, time.Second) require.Empty(t, source) // And so is no host at all, which is what an un-plumbed policy looks // like. nohost := hinted nohost.Host = "" _, source = nohost.delay(1, errors.New("boom")) require.Empty(t, source) }) t.Run("the further-out of the two sources wins", func(t *testing.T) { hints := NewBackoffHints() hints.Observe("pds.example", http.StatusTooManyRequests, http.Header{"Retry-After": []string{"2"}}) hinted := RetryPolicy{BaseDelay: time.Second, MaxDelay: 30 * time.Second, Hints: hints, Host: "pds.example"} // indigo parsed a reset further out than the header we captured. d, source := hinted.delay(1, ratelimited(time.Now().Add(6*time.Second))) require.Greater(t, d, 5*time.Second) require.Equal(t, "ratelimit-reset", source) // And the other way around. d, source = hinted.delay(1, ratelimited(time.Now().Add(time.Millisecond))) require.Greater(t, d, 1500*time.Millisecond) require.Equal(t, "retry-after", source) }) t.Run("a hint is clamped to MaxDelay", func(t *testing.T) { hints := NewBackoffHints() hints.Observe("pds.example", http.StatusTooManyRequests, http.Header{"Retry-After": []string{"600"}}) hinted := RetryPolicy{BaseDelay: time.Second, MaxDelay: 30 * time.Second, Hints: hints, Host: "pds.example"} d, source := hinted.delay(1, errors.New("boom")) require.Equal(t, 30*time.Second, d) require.Equal(t, "retry-after", source) }) t.Run("a stale observation does not inflate a later wait", func(t *testing.T) { hints := NewBackoffHints() // An hour-long backoff, observed longer ago than hintMaxAge: the wait it // asked for has not elapsed, but it is no longer evidence about now. hints.observeAt("pds.example", http.StatusTooManyRequests, http.Header{"Retry-After": []string{"3600"}}, time.Now().Add(-hintMaxAge-time.Minute)) hinted := RetryPolicy{BaseDelay: time.Second, MaxDelay: 30 * time.Second, Hints: hints, Host: "pds.example"} d, source := hinted.delay(1, errors.New("boom")) require.LessOrEqual(t, d, time.Second) require.Empty(t, source) _, live := hints.Get("pds.example") require.False(t, live) }) } // nxdomain is a name that no longer resolves, wrapped the way it arrives: the // resolver's error inside net/http's dial error inside net/http's request // error. func nxdomain() error { return fmt.Errorf("getBlocks: %w", &url.Error{ Op: "Get", URL: "https://gone.example/xrpc/com.atproto.sync.getBlocks", Err: &net.OpError{Op: "dial", Net: "tcp", Err: &net.DNSError{ Err: "no such host", Name: "gone.example", IsNotFound: true, }}, }) } // TestRetryFastFailsDeadHosts: a host that is not there at all gets two // attempts, not five. This is what makes a sweep's straggler tail cheap -- // switched-off PDSes were costing a full backoff ladder per repo to rediscover // something the first connection attempt already reported. func TestRetryFastFailsDeadHosts(t *testing.T) { policy := RetryPolicy{MaxAttempts: 5, BaseDelay: time.Millisecond, MaxDelay: 2 * time.Millisecond} attempts := func(t *testing.T, err error) int { t.Helper() calls := 0 got := policy.do(context.Background(), "getBlocks", func() error { calls++ return err }) require.Error(t, got) return calls } t.Run("connection refused", func(t *testing.T) { err := fmt.Errorf("request failed: %w", &url.Error{Op: "Get", URL: "https://pds.example/", Err: &net.OpError{Op: "dial", Err: syscall.ECONNREFUSED}}) require.True(t, isDeadHost(err)) // Still retryable: a PDS that is restarting refuses for a moment. require.True(t, isRetryable(err)) require.Equal(t, deadHostAttempts, attempts(t, err)) }) t.Run("no such host", func(t *testing.T) { err := nxdomain() require.True(t, isDeadHost(err)) // A name that does not resolve now will not resolve in a second, so // this never even reaches the two-attempt cap: it is not retryable at // all, and one attempt is what it costs. require.False(t, isRetryable(err)) require.Equal(t, 1, attempts(t, err)) }) t.Run("a timeout still gets the whole ladder", func(t *testing.T) { // The host is there and answering slowly, which is exactly what the // retries are for. err := fmt.Errorf("request failed: %w", timeoutError{}) require.False(t, isDeadHost(err)) require.Equal(t, 5, attempts(t, err)) }) t.Run("a DNS timeout is not a dead host", func(t *testing.T) { // The resolver is struggling, not answering "no": that is transient. err := fmt.Errorf("request failed: %w", &net.OpError{Op: "dial", Err: &net.DNSError{ Err: "i/o timeout", Name: "pds.example", IsTimeout: true, }}) require.False(t, isDeadHost(err)) require.Equal(t, 5, attempts(t, err)) }) t.Run("429 still gets the whole ladder", func(t *testing.T) { err := ratelimited(time.Time{}) require.False(t, isDeadHost(err)) require.Equal(t, 5, attempts(t, err)) }) t.Run("503 still gets the whole ladder", func(t *testing.T) { require.Equal(t, 5, attempts(t, xrpcErr(http.StatusServiceUnavailable, "", "restarting"))) }) t.Run("a policy that asks for less keeps it", func(t *testing.T) { single := RetryPolicy{MaxAttempts: 1, BaseDelay: time.Millisecond} calls := 0 err := single.do(context.Background(), "getBlocks", func() error { calls++ return fmt.Errorf("dialing: %w", syscall.ECONNREFUSED) }) require.Error(t, err) require.Equal(t, 1, calls) }) } // TestRetryDeadHostOffTheWire: the classification above is only worth anything // if a refused connection still looks like one after net/http and indigo have // each wrapped it, so this one dials a port that nothing is listening on. func TestRetryDeadHostOffTheWire(t *testing.T) { ln, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) addr := ln.Addr().String() require.NoError(t, ln.Close()) sr := buildSignedRepo(t, testDID, exactnessPaths()) f := &XRPCBlockFetcher{ Client: &xrpc.Client{Host: "http://" + addr}, DID: testDID, Retry: RetryPolicy{MaxAttempts: 5, BaseDelay: time.Millisecond, MaxDelay: 2 * time.Millisecond}, } _, err = f.GetBlocks(context.Background(), []cid.Cid{sr.root}) require.Error(t, err) require.True(t, isRetryable(err), "a refused connection is worth one retry: %v", err) require.True(t, isDeadHost(err), "but it must be recognisable as a dead host: %v", err) require.Contains(t, err.Error(), fmt.Sprintf("giving up after %d attempts", deadHostAttempts)) } func ratelimited(reset time.Time) error { return fmt.Errorf("getBlocks: %w", &xrpc.Error{ StatusCode: http.StatusTooManyRequests, Wrapped: &xrpc.XRPCError{ErrStr: "RateLimitExceeded"}, Ratelimit: &xrpc.RatelimitInfo{Limit: 3000, Reset: reset}, }) } // TestErrForLog: the retry warning is the one line an operator sees when a host // is throttling a walk, and for the commonest case -- a 429 with an HTML body -- // indigo's JSON decoder failure was burying the status code in parser trivia. func TestErrForLog(t *testing.T) { // The genuine article, straight off the wire. sr := buildSignedRepo(t, testDID, exactnessPaths()) host := newFakeHost(sr) host.blocksFailures = []failure{htmlThrottled} client := host.start(t) f := &XRPCBlockFetcher{Client: client, DID: testDID, Retry: RetryPolicy{MaxAttempts: 1}} _, err := f.GetBlocks(context.Background(), []cid.Cid{sr.root}) require.Error(t, err) require.Contains(t, err.Error(), "failed to decode xrpc error message", "fixture no longer produces the shape under test") require.Equal(t, "HTTP 429 (undecodable error body)", errForLog(err)) // A host that answered properly keeps its whole message. proper := xrpcErr(http.StatusTooManyRequests, "RateLimitExceeded", "Rate Limit Exceeded") require.Equal(t, proper, errForLog(proper)) // So does anything that never reached a host. plain := fmt.Errorf("dialing: %w", syscall.ECONNREFUSED) require.Equal(t, plain, errForLog(plain)) // And a 502 from a load balancer gets the same treatment as the 429. gateway := fmt.Errorf("getBlocks: %w", &xrpc.Error{ StatusCode: http.StatusBadGateway, Wrapped: fmt.Errorf("failed to decode xrpc error message: %w", errors.New("invalid character '<'")), }) require.Equal(t, "HTTP 502 (undecodable error body)", errForLog(gateway)) } // TestXRPCBlockFetcherRetries drives the retry loop through the real getBlocks // path against an HTTP host that fails on a script. func TestXRPCBlockFetcherRetries(t *testing.T) { ctx := context.Background() sr := buildSignedRepo(t, testDID, exactnessPaths()) newFetcher := func(t *testing.T, script ...failure) (*XRPCBlockFetcher, *fakeHost) { host := newFakeHost(sr) host.blocksFailures = script client := host.start(t) return &XRPCBlockFetcher{Client: client, DID: testDID, Retry: fastRetry()}, host } t.Run("429 then success", func(t *testing.T) { f, host := newFetcher(t, throttled(time.Time{}), throttled(time.Time{})) blocks, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.NoError(t, err) require.Contains(t, blocks, sr.root) require.Equal(t, 3, host.requests, "two throttles then the real answer") }) t.Run("429 with an HTML body then success", func(t *testing.T) { // The production shape: indigo cannot decode the body, so the error is // "failed to decode xrpc error message: invalid character '<'" and only // the status code is left to classify on. f, host := newFetcher(t, htmlThrottled) _, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.NoError(t, err) require.Equal(t, 2, host.requests) }) t.Run("503 then success", func(t *testing.T) { f, host := newFetcher(t, failure{status: http.StatusServiceUnavailable, body: "upstream restarting"}) _, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.NoError(t, err) require.Equal(t, 2, host.requests) }) t.Run("a ratelimit reset in the far future is capped, not slept on", func(t *testing.T) { f, host := newFetcher(t, throttled(time.Now().Add(5*time.Minute))) start := time.Now() _, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.NoError(t, err) require.Equal(t, 2, host.requests) require.Less(t, time.Since(start), time.Second, "MaxDelay must bound the ratelimit wait") }) t.Run("400 fails fast", func(t *testing.T) { // What a TS PDS says when the walk raced a live repo and the blocks of // the pinned commit have been garbage collected. Retrying the same CIDs // can never help. f, host := newFetcher(t, failure{ status: http.StatusBadRequest, body: `{"error":"InvalidRequest","message":"Could not find cids: bafyreib2"}`, header: map[string]string{"Content-Type": "application/json"}, }) _, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.Error(t, err) require.Equal(t, 1, host.requests, "no retries") var xe *xrpc.Error require.ErrorAs(t, err, &xe) require.Equal(t, http.StatusBadRequest, xe.StatusCode) require.NotContains(t, err.Error(), "giving up", "a fail-fast error is passed through unchanged") }) t.Run("retries exhausted", func(t *testing.T) { f, host := newFetcher(t, throttled(time.Time{}), throttled(time.Time{}), throttled(time.Time{}), throttled(time.Time{}), throttled(time.Time{}), throttled(time.Time{})) _, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.Error(t, err) require.Equal(t, 5, host.requests, "MaxAttempts is a total, not an extra") require.Contains(t, err.Error(), "giving up after 5 attempts") var xe *xrpc.Error require.ErrorAs(t, err, &xe, "the last error is still inspectable") require.Equal(t, http.StatusTooManyRequests, xe.StatusCode) }) t.Run("context cancelled mid backoff", func(t *testing.T) { host := newFakeHost(sr) host.blocksFailures = []failure{throttled(time.Time{})} client := host.start(t) // A backoff long enough that returning promptly can only mean the // sleep was context aware. f := &XRPCBlockFetcher{Client: client, DID: testDID, Retry: RetryPolicy{MaxAttempts: 5, BaseDelay: 30 * time.Second, MaxDelay: time.Minute}} ctx, cancel := context.WithCancel(context.Background()) go func() { time.Sleep(20 * time.Millisecond) cancel() }() start := time.Now() _, err := f.GetBlocks(ctx, []cid.Cid{sr.root}) require.Error(t, err) require.Less(t, time.Since(start), 5*time.Second) require.ErrorIs(t, err, context.Canceled) var xe *xrpc.Error require.ErrorAs(t, err, &xe, "the failure that triggered the backoff is kept too") require.Equal(t, http.StatusTooManyRequests, xe.StatusCode) require.Equal(t, 1, host.requests) }) t.Run("retrying does not break chunking", func(t *testing.T) { // A retry inside one chunk must not disturb the chunk loop: five CIDs // at ChunkSize 2 is three chunks, and the failure only costs one extra // call. host := newFakeHost(sr) host.blocksFailures = []failure{throttled(time.Time{})} client := host.start(t) f := &XRPCBlockFetcher{Client: client, DID: testDID, ChunkSize: 2, Retry: fastRetry()} want := []cid.Cid{sr.root, sr.commitCID} for c := range sr.blocks { if len(want) >= 5 { break } if c != sr.root && c != sr.commitCID { want = append(want, c) } } require.Len(t, want, 5) blocks, err := f.GetBlocks(ctx, want) require.NoError(t, err) require.Len(t, blocks, len(want)) require.Equal(t, 4, host.requests, "three chunks plus the one retry") }) } // TestFetchVerifiedHeadRetries: the head fetch is one getLatestCommit call, and // it is the first thing every backfill does, so it gets the same treatment. func TestFetchVerifiedHeadRetries(t *testing.T) { ctx := context.Background() sr := buildSignedRepo(t, testDID, exactnessPaths()) pub, err := sr.priv.PublicKey() require.NoError(t, err) t.Run("throttled then success", func(t *testing.T) { host := newFakeHost(sr) host.latestFailures = []failure{throttled(time.Time{}), htmlThrottled, {status: http.StatusServiceUnavailable, body: "restarting"}} client := host.start(t) f := &XRPCBlockFetcher{Client: client, DID: testDID, Retry: fastRetry()} head, err := FetchVerifiedHead(ctx, client, f, sr.directory(t, testDID, pub), testDID, fastRetry()) require.NoError(t, err) require.Equal(t, sr.commitCID, head.CID) require.Equal(t, 4, host.latestRequests) }) t.Run("RepoNotFound fails fast", func(t *testing.T) { host := newFakeHost(sr) host.latestFailures = []failure{{ status: http.StatusBadRequest, body: `{"error":"RepoNotFound","message":"could not find repo"}`, header: map[string]string{"Content-Type": "application/json"}, }} client := host.start(t) f := &XRPCBlockFetcher{Client: client, DID: testDID, Retry: fastRetry()} _, err := FetchVerifiedHead(ctx, client, f, sr.directory(t, testDID, pub), testDID, fastRetry()) require.Error(t, err) require.Equal(t, 1, host.latestRequests) }) t.Run("at most one policy", func(t *testing.T) { host := newFakeHost(sr) client := host.start(t) f := &XRPCBlockFetcher{Client: client, DID: testDID} _, err := FetchVerifiedHead(ctx, client, f, sr.directory(t, testDID, pub), testDID, fastRetry(), fastRetry()) require.Error(t, err) }) }