diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index b96a726..79e18fa 100644 --- a/pkg/hold/gc/gc.go +++ b/pkg/hold/gc/gc.go @@ -3,15 +3,19 @@ package gc import ( "bytes" "context" + "crypto/tls" "encoding/json" + "errors" "fmt" "io" "log/slog" + "net" "net/http" "net/url" "regexp" "strings" "sync" + "syscall" "time" "atcr.io/pkg/atproto" @@ -1660,6 +1664,122 @@ func (gc *GarbageCollector) fetchUserProfile(ctx context.Context, pdsEndpoint, u return &profile, nil } +// GC walks a user's PDS to decide which blobs are still referenced, and a +// failed walk is treated as "assume everything is referenced" — safe, but it +// pins that user's storage for the whole run. Finding 16 measured the cost: +// every DID GC classified unreachable-but-healthy had failed only one or two +// runs out of six, and replaying the identical calls afterwards returned 200 +// in 45-680 ms with no rate limiting. These are ordinary blips on small +// self-hosted PDSes, amplified into a full-DID skip by a single-shot fetch. +const ( + gcListAttempts = 3 + gcListBackoff = 500 * time.Millisecond +) + +// retryableFetchErr reports whether a transport error is worth another attempt. +// The distinction matters in both directions: retrying a blip recovers a +// healthy user's storage, but retrying a host that is genuinely gone only slows +// the run and makes a dead PDS look alive for longer. DNS failure, TLS failure +// and connection refused are stable facts about an endpoint and repeat +// identically, so they are never retried. Timeouts and resets are the transient +// shapes. Anything unrecognised is treated as permanent, which errs toward the +// current behaviour rather than toward hammering. +func retryableFetchErr(err error) bool { + if err == nil { + return false + } + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return false + } + var certErr *tls.CertificateVerificationError + if errors.As(err, &certErr) { + return false + } + var recordErr tls.RecordHeaderError + if errors.As(err, &recordErr) { + return false + } + if errors.Is(err, syscall.ECONNREFUSED) { + return false + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return true + } + return errors.Is(err, syscall.ECONNRESET) || errors.Is(err, io.ErrUnexpectedEOF) +} + +// retryableStatus reports whether an HTTP status deserves another attempt. +// 5xx and 429 are the server saying "not now"; a 4xx is the server saying +// "not ever", and repeating it would be pointless. +func retryableStatus(status int) bool { + return status == http.StatusTooManyRequests || status >= 500 +} + +// listRecordsPage performs one listRecords GET and decodes it into out, +// retrying transient failures with linear backoff. The caller's context still +// short-circuits everything: a cancelled run stops immediately rather than +// sleeping through its remaining attempts. +func (gc *GarbageCollector) listRecordsPage(ctx context.Context, client *http.Client, reqURL string, out any) error { + var lastErr error + + for attempt := 1; attempt <= gcListAttempts; attempt++ { + if attempt > 1 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Duration(attempt-1) * gcListBackoff): + } + } + + req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + + resp, err := client.Do(req) + if err != nil { + lastErr = fmt.Errorf("http request: %w", err) + if ctx.Err() != nil || !retryableFetchErr(err) { + return lastErr + } + slog.Debug("GC listRecords attempt failed, retrying", + "component", "gc", "url", reqURL, "attempt", attempt, "error", err) + continue + } + + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + lastErr = fmt.Errorf("listRecords returned status %d", resp.StatusCode) + if !retryableStatus(resp.StatusCode) { + return lastErr + } + slog.Debug("GC listRecords attempt returned retryable status", + "component", "gc", "url", reqURL, "attempt", attempt, "status", resp.StatusCode) + continue + } + + err = json.NewDecoder(resp.Body).Decode(out) + resp.Body.Close() + if err != nil { + // A truncated body is a transport-shaped failure, so it is worth + // one more attempt rather than writing the whole DID off. + lastErr = fmt.Errorf("decode response: %w", err) + if ctx.Err() != nil || !retryableFetchErr(err) { + return lastErr + } + slog.Debug("GC listRecords decode failed, retrying", + "component", "gc", "url", reqURL, "attempt", attempt, "error", err) + continue + } + + return nil + } + + return fmt.Errorf("after %d attempts: %w", gcListAttempts, lastErr) +} + // fetchUserTags fetches all tag records from a user's PDS. // Returns a map of digest → true for all tagged manifest digests. func (gc *GarbageCollector) fetchUserTags(ctx context.Context, pdsEndpoint, userDID string) (map[string]bool, error) { @@ -1676,21 +1796,6 @@ func (gc *GarbageCollector) fetchUserTags(ctx context.Context, pdsEndpoint, user reqURL += "&cursor=" + cursor } - req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) - if err != nil { - return nil, fmt.Errorf("create request: %w", err) - } - - resp, err := client.Do(req) - if err != nil { - return nil, fmt.Errorf("http request: %w", err) - } - - if resp.StatusCode != http.StatusOK { - resp.Body.Close() - return nil, fmt.Errorf("listRecords returned status %d", resp.StatusCode) - } - var listResult struct { Records []struct { Value json.RawMessage `json:"value"` @@ -1698,11 +1803,9 @@ func (gc *GarbageCollector) fetchUserTags(ctx context.Context, pdsEndpoint, user Cursor string `json:"cursor,omitempty"` } - if err := json.NewDecoder(resp.Body).Decode(&listResult); err != nil { - resp.Body.Close() - return nil, fmt.Errorf("decode response: %w", err) + if err := gc.listRecordsPage(ctx, client, reqURL, &listResult); err != nil { + return nil, err } - resp.Body.Close() for _, rec := range listResult.Records { var tag atproto.TagRecord @@ -1782,21 +1885,6 @@ func (gc *GarbageCollector) fetchUserManifestsFromEndpoint(ctx context.Context, reqURL += "&cursor=" + cursor } - req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) - if err != nil { - return nil, fmt.Errorf("create request: %w", err) - } - - resp, err := client.Do(req) - if err != nil { - return nil, fmt.Errorf("http request: %w", err) - } - - if resp.StatusCode != http.StatusOK { - resp.Body.Close() - return nil, fmt.Errorf("listRecords returned status %d", resp.StatusCode) - } - var listResult struct { Records []struct { URI string `json:"uri"` @@ -1806,11 +1894,9 @@ func (gc *GarbageCollector) fetchUserManifestsFromEndpoint(ctx context.Context, Cursor string `json:"cursor,omitempty"` } - if err := json.NewDecoder(resp.Body).Decode(&listResult); err != nil { - resp.Body.Close() - return nil, fmt.Errorf("decode response: %w", err) + if err := gc.listRecordsPage(ctx, client, reqURL, &listResult); err != nil { + return nil, err } - resp.Body.Close() for _, rec := range listResult.Records { var manifest atproto.ManifestRecord diff --git a/pkg/hold/gc/list_retry_test.go b/pkg/hold/gc/list_retry_test.go new file mode 100644 index 0000000..506853b --- /dev/null +++ b/pkg/hold/gc/list_retry_test.go @@ -0,0 +1,155 @@ +package gc + +import ( + "context" + "errors" + "net" + "net/http" + "net/http/httptest" + "syscall" + "testing" + "time" +) + +// A transient blip must not cost a user their whole GC run. Finding 16: a +// single-shot fetch turned one failed page into "assume everything is +// referenced" for that DID, pinning healthy users' storage for the entire run. +func TestListRecordsPage_RetriesTransientFailures(t *testing.T) { + tests := []struct { + name string + failures int + failStatus int + wantAttempts int + wantErr bool + }{ + {"succeeds first try", 0, 0, 1, false}, + {"recovers after one 503", 1, http.StatusServiceUnavailable, 2, false}, + {"recovers after two 500s", 2, http.StatusInternalServerError, 3, false}, + {"recovers after 429", 1, http.StatusTooManyRequests, 2, false}, + {"gives up after the attempt budget", 5, http.StatusBadGateway, gcListAttempts, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var attempts int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempts++ + if attempts <= tt.failures { + w.WriteHeader(tt.failStatus) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"records":[],"cursor":""}`)) + })) + defer srv.Close() + + gc := &GarbageCollector{} + var out struct { + Cursor string `json:"cursor"` + } + err := gc.listRecordsPage(context.Background(), srv.Client(), srv.URL, &out) + + if tt.wantErr && err == nil { + t.Fatal("expected an error once the attempt budget is spent") + } + if !tt.wantErr && err != nil { + t.Fatalf("expected recovery, got: %v", err) + } + if attempts != tt.wantAttempts { + t.Errorf("made %d attempts, want %d", attempts, tt.wantAttempts) + } + }) + } +} + +// A 4xx is the server saying "not ever". Retrying it wastes the run and, worse, +// keeps a genuinely dead PDS looking alive for longer. +func TestListRecordsPage_DoesNotRetryPermanentStatus(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusNotFound} { + var attempts int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempts++ + w.WriteHeader(status) + })) + + gc := &GarbageCollector{} + var out struct{} + err := gc.listRecordsPage(context.Background(), srv.Client(), srv.URL, &out) + srv.Close() + + if err == nil { + t.Errorf("status %d: expected an error", status) + } + if attempts != 1 { + t.Errorf("status %d: made %d attempts, want exactly 1", status, attempts) + } + } +} + +// A cancelled run must stop immediately rather than sleeping through its +// remaining backoff. +func TestListRecordsPage_HonoursContextCancellation(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + gc := &GarbageCollector{} + var out struct{} + + start := time.Now() + if err := gc.listRecordsPage(ctx, srv.Client(), srv.URL, &out); err == nil { + t.Fatal("expected an error for a cancelled context") + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("took %v, should have bailed out immediately", elapsed) + } +} + +// The classification in finding 16 only holds if "transient" excludes the +// failures that are really facts about an endpoint. Getting this wrong the +// permissive way makes a dead PDS look alive; getting it wrong the strict way +// just restores today's behaviour. +func TestRetryableFetchErr(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"DNS failure is a fact about the endpoint", &net.DNSError{Err: "no such host", IsNotFound: true}, false}, + {"connection refused is a fact about the endpoint", syscall.ECONNREFUSED, false}, + {"connection reset is a blip", syscall.ECONNRESET, true}, + // A DNS timeout is still a DNS failure: excluded before the timeout + // check, deliberately, so a resolver hiccup does not make a dead host + // look retryable. + {"DNS timeout is still a DNS failure", &net.DNSError{Err: "timeout", IsTimeout: true}, false}, + {"unrecognised errors stay permanent", errors.New("something else"), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := retryableFetchErr(tt.err); got != tt.want { + t.Errorf("retryableFetchErr(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +func TestRetryableStatus(t *testing.T) { + retryable := []int{http.StatusTooManyRequests, http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable} + permanent := []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound, http.StatusGone} + + for _, s := range retryable { + if !retryableStatus(s) { + t.Errorf("status %d should be retryable", s) + } + } + for _, s := range permanent { + if retryableStatus(s) { + t.Errorf("status %d should not be retryable", s) + } + } +}