diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index 8d86ee7..f4a1d92 100644 --- a/pkg/hold/gc/gc.go +++ b/pkg/hold/gc/gc.go @@ -249,7 +249,19 @@ type GarbageCollector struct { // predecessorCache caches holdDID → "is this a predecessor of our hold?" // A predecessor is a hold whose captain record has a successor label set. + // + // Only definitive answers belong here. The cache is never reset, so a false + // recorded from an unreachable hold would outlive the outage and unreference + // that hold's blobs on every subsequent run — deleting content this hold is + // still serving on the predecessor's behalf. predecessorCache map[string]bool + + // predecessorUnresolved holds the DIDs whose predecessor status could not be + // determined during the current analysis. It exists only so that one + // unreachable hold costs a single 5s timeout per run rather than one per + // manifest, and it is cleared at the start of every analysis so a hold that + // was down once is re-checked next time instead of being written off. + predecessorUnresolved map[string]bool } // GCResult contains statistics from a GC run @@ -289,12 +301,13 @@ type analysisResult struct { // the labeler-aware takedown gate) is configured via Option arguments. func NewGarbageCollector(holdPDS *pds.HoldPDS, s3svc *s3.S3Service, cfg Config, opts ...Option) *GarbageCollector { gc := &GarbageCollector{ - pds: holdPDS, - s3: s3svc, - cfg: cfg, - logger: slog.Default().With("component", "gc"), - stopCh: make(chan struct{}), - predecessorCache: make(map[string]bool), + pds: holdPDS, + s3: s3svc, + cfg: cfg, + logger: slog.Default().With("component", "gc"), + stopCh: make(chan struct{}), + predecessorCache: make(map[string]bool), + predecessorUnresolved: make(map[string]bool), } for _, opt := range opts { opt(gc) @@ -742,6 +755,11 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult referenced: make(map[string]bool), } + // Start each analysis with a clean slate of unresolved holds, so a hold that + // was unreachable last night gets another chance tonight. Definitive answers + // in predecessorCache are kept — those do not go stale within a process. + gc.predecessorUnresolved = make(map[string]bool) + recordsIndex := gc.pds.RecordsIndex() if recordsIndex == nil { return nil, fmt.Errorf("records index not available") @@ -1789,7 +1807,10 @@ func (gc *GarbageCollector) manifestBelongsToHold(ctx context.Context, manifest // isPredecessorHold checks if the given holdDID is a predecessor of this hold // by fetching its captain record and checking for a successor label. -// Results are cached to avoid repeated network calls. +// Results are cached to avoid repeated network calls. An inconclusive check +// reports true — the manifest is kept referenced — matching how this package +// already treats a user whose PDS cannot be reached: an outage must never be +// the reason content becomes eligible for deletion. func (gc *GarbageCollector) isPredecessorHold(ctx context.Context, holdDID string) bool { if gc.predecessorCache == nil { gc.predecessorCache = make(map[string]bool) @@ -1799,14 +1820,41 @@ func (gc *GarbageCollector) isPredecessorHold(ctx context.Context, holdDID strin return isPredecessor } - isPredecessor := gc.checkPredecessor(ctx, holdDID) + // Already inconclusive earlier in this run. Answer the same way without + // paying another timeout; the next run starts fresh and re-checks. + if gc.predecessorUnresolved[holdDID] { + return true + } + + isPredecessor, definitive := gc.checkPredecessor(ctx, holdDID) + if !definitive { + if gc.predecessorUnresolved == nil { + gc.predecessorUnresolved = make(map[string]bool) + } + gc.predecessorUnresolved[holdDID] = true + gc.logger.Warn("GC: predecessor status unresolved, treating hold's manifests as referenced", + "holdDID", holdDID) + return true + } + gc.predecessorCache[holdDID] = isPredecessor return isPredecessor } // checkPredecessor fetches a hold's captain record to check if it has a successor label // (meaning the hold has been migrated/retired and its blobs are served by this hold). -func (gc *GarbageCollector) checkPredecessor(ctx context.Context, holdDID string) bool { +// +// The second return value reports whether the answer is definitive. It is false whenever +// the hold could not be reached or its reply could not be understood: those are cases +// where the hold may well be a predecessor and we simply cannot tell. Callers must not +// read an inconclusive result as "not a predecessor" — a single five-second blip would +// otherwise drop a live predecessor's entire blob set out of the referenced set, and +// those blobs are long past the grace period that protects recent content. +// +// A non-200 is treated as inconclusive rather than a negative on the same reasoning: a +// reachable service that cannot produce its own captain record is malfunctioning, not +// answering. Over-protecting an unrelated hold merely leaves some blobs unreclaimed. +func (gc *GarbageCollector) checkPredecessor(ctx context.Context, holdDID string) (isPredecessor, definitive bool) { fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() @@ -1814,56 +1862,74 @@ func (gc *GarbageCollector) checkPredecessor(ctx context.Context, holdDID string if err != nil { gc.logger.Debug("GC: failed to resolve predecessor hold URL", "holdDID", holdDID, "error", err) - return false + return false, false } + return gc.checkPredecessorAt(fetchCtx, holdDID, holdURL) +} + +// checkPredecessorAt is checkPredecessor with the hold's base URL already resolved, +// split out so the fetch-and-parse half can be exercised against a local server. +// It carries the same contract: the second return value is false whenever the answer +// is inconclusive rather than negative. +func (gc *GarbageCollector) checkPredecessorAt(ctx context.Context, holdDID, holdURL string) (isPredecessor, definitive bool) { recordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self", holdURL, url.QueryEscape(holdDID), url.QueryEscape(atproto.CaptainCollection), ) - req, err := http.NewRequestWithContext(fetchCtx, "GET", recordURL, nil) + req, err := http.NewRequestWithContext(ctx, "GET", recordURL, nil) if err != nil { - return false + return false, false } resp, err := http.DefaultClient.Do(req) if err != nil { gc.logger.Debug("GC: failed to fetch predecessor captain record", "holdDID", holdDID, "error", err) - return false + return false, false } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return false + gc.logger.Debug("GC: predecessor captain record fetch returned non-200", + "holdDID", holdDID, "status", resp.StatusCode) + return false, false } body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) if err != nil { - return false + gc.logger.Debug("GC: failed to read predecessor captain record", + "holdDID", holdDID, "error", err) + return false, false } var envelope struct { Value json.RawMessage `json:"value"` } if err := json.Unmarshal(body, &envelope); err != nil { - return false + gc.logger.Debug("GC: failed to parse predecessor captain envelope", + "holdDID", holdDID, "error", err) + return false, false } var captain atproto.CaptainRecord if err := json.Unmarshal(envelope.Value, &captain); err != nil { - return false + gc.logger.Debug("GC: failed to parse predecessor captain record", + "holdDID", holdDID, "error", err) + return false, false } if captain.Successor != "" { gc.logger.Info("GC: discovered predecessor hold (has successor label)", "holdDID", holdDID, "successor", captain.Successor) - return true + return true, true } - return false + // The hold answered and declares no successor. This is the one negative we + // are entitled to cache. + return false, true } // deleteOrphanedRecords removes layer records whose manifests no longer exist diff --git a/pkg/hold/gc/gc_test.go b/pkg/hold/gc/gc_test.go index 6503e4e..a173f15 100644 --- a/pkg/hold/gc/gc_test.go +++ b/pkg/hold/gc/gc_test.go @@ -201,7 +201,16 @@ func TestConfig(t *testing.T) { } func TestManifestBelongsToHold(t *testing.T) { - gc := &GarbageCollector{logger: newTestLogger()} + gc := &GarbageCollector{ + logger: newTestLogger(), + // Seeded with a definitive answer so the non-matching cases resolve from + // cache instead of the network: this test is about routing, not + // reachability. An unreachable hold deliberately answers the other way, + // covered by TestPredecessorUnresolvedStaysReferenced. + predecessorCache: map[string]bool{ + "did:web:other-hold.atcr.io": false, + }, + } holdDID := "did:web:hold01.atcr.io" tests := []struct { @@ -262,6 +271,57 @@ func TestManifestBelongsToHold(t *testing.T) { } } +// A hold that cannot be reached must never be written off as "not a predecessor". +// Doing so drops a live predecessor's entire blob set out of the referenced set, +// and those blobs are long past the grace period that protects recent content, so +// the next run deletes them outright. The inconclusive answer must also stay out +// of predecessorCache, which is never reset: caching it would make a single +// five-second blip permanent for the life of the process. +func TestPredecessorUnresolvedStaysReferenced(t *testing.T) { + gc := &GarbageCollector{logger: newTestLogger()} + holdDID := "did:web:hold01.atcr.io" + + // .invalid is reserved by RFC 2606 and never resolves. + const unreachable = "did:web:unreachable.invalid" + manifest := &atproto.ManifestRecord{HoldDID: unreachable} + + if !gc.manifestBelongsToHold(context.Background(), manifest, holdDID) { + t.Fatal("unreachable hold treated as not-a-predecessor: its blobs would be deleted") + } + + if _, cached := gc.predecessorCache[unreachable]; cached { + t.Error("inconclusive check cached in predecessorCache: the outage would outlive itself") + } + + if !gc.predecessorUnresolved[unreachable] { + t.Error("inconclusive check not recorded in predecessorUnresolved: it would re-dial per manifest") + } + + // Second call must be answered from predecessorUnresolved, not re-dialled. + if !gc.manifestBelongsToHold(context.Background(), manifest, holdDID) { + t.Error("second call disagreed with the first") + } +} + +// A hold that answers and declares no successor is the one negative worth caching. +func TestPredecessorDefinitiveNegativeIsCached(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"uri":"at://x","value":{"successor":""}}`)) + })) + defer srv.Close() + + gc := &GarbageCollector{logger: newTestLogger()} + isPredecessor, definitive := gc.checkPredecessorAt(context.Background(), "did:web:example.test", srv.URL) + + if isPredecessor { + t.Error("hold with no successor reported as a predecessor") + } + if !definitive { + t.Error("a 200 with a parseable captain record should be definitive") + } +} + func TestFetchUserManifests(t *testing.T) { holdDID := "did:web:hold.example.com" @@ -333,6 +393,14 @@ func TestFetchUserManifests(t *testing.T) { gc := &GarbageCollector{ pds: nil, // Not used directly in fetchUserManifests when we bypass DID resolution logger: newTestLogger(), + // The other hold is seeded as a definitive non-predecessor. Without this + // it would be unreachable rather than negative, and an unreachable hold + // is deliberately kept referenced (see + // TestPredecessorUnresolvedStaysReferenced), which is not what this test + // is about. + predecessorCache: map[string]bool{ + "did:web:other-hold.example.com": false, + }, } // Call fetchUserManifestsFromEndpoint (bypasses DID resolution)