gc: retry transient PDS failures instead of pinning a user's storage

Both paginated walks bailed on the first error, and the caller treats a failed walk as "assume everything is referenced", so one blip skipped that DID's storage for the whole run. Measured before changing anything: every DID GC had classified unreachable but healthy failed only one or two runs out of six, and replaying the exact same listRecords calls afterwards returned 200 in 45-680 ms with no rate limiting. Ordinary blips on small self-hosted PDSes, amplified into a full-DID skip. Share one listRecordsPage helper between fetchUserTags and fetchUserManifestsFromEndpoint. The retry decision splits deliberately: timeouts, connection reset, 5xx and 429 get another attempt, while DNS failure, TLS failure, connection refused and any 4xx do not. Those are stable facts about an endpoint, and retrying them would only slow the run and keep a dead PDS looking alive longer. Unrecognised errors stay permanent, so an unfamiliar failure degrades to today's behaviour rather than hammering. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UAqi2hS2dhZoatqcWoYZQk
2026-09-03 16:56:56 +00:00 · 2026-09-02 12:44:33 -05:00
parent c36e90f6b7
commit a4aedbd2c8
2 changed files with 279 additions and 38 deletions
@@ -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
@@ -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)
+		}
+	}
+}