hold/gc: require a predecessor's successor to name this hold

2e55352 taught the scan broadcaster that a successor label is only
interesting when it points at us; the GC copy of the same check was left
accepting any non-empty successor. Confirmed still divergent at the head of
this stack: gc.go was a bare `if captain.Successor != ""` while
scan_broadcaster.go:1473 compares against sb.holdDID.

A hold that retired into some third hold is that hold's predecessor, not
ours, and its manifests are not a reason to keep blobs referenced here. GC
therefore now makes the same comparison the broadcaster does, against
gc.pds.DID().

This is the one change in the batch that makes GC delete more rather than
less, so it is deliberately its own commit and carries a floor. If this hold
cannot say who it is, ourHoldDID() returns "" and the old permissive answer
stands: we cannot conclude a successor is not us, and over-protecting merely
leaks blobs while guessing the other way destroys them. That branch has its
own test, because an empty DID silently turning every predecessor into a
stranger is exactly how this reconciliation would become the next blob-loss
bug.

The inconclusive-on-failure semantics 95d4f7c added are untouched: only the
answers from a hold that actually replied are affected.

Verified by mutation: forcing the comparison back to the permissive form
fails exactly one case, the successor naming a third hold, and leaves the
unknown-own-DID fallback passing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/hold/gc/gc.go

@@ -1921,15 +1921,44 @@ func (gc *GarbageCollector) checkPredecessorAt(ctx context.Context, holdDID, hol
 		return false, false
 	}
 
-	if captain.Successor != "" {
-		gc.logger.Info("GC: discovered predecessor hold (has successor label)",
+	// The hold answered and declares no successor. This is the one negative we
+	// are entitled to cache.
+	if captain.Successor == "" {
+		return false, true
+	}
+
+	// A successor label alone is not enough: it has to name us. A hold that
+	// retired into some third hold is that hold's predecessor, not ours, and its
+	// manifests are not a reason to keep blobs referenced here.
+	// scan_broadcaster.go makes exactly this comparison, and the two are meant
+	// to agree.
+	ourHoldDID := gc.ourHoldDID()
+	if ourHoldDID == "" {
+		// We cannot say who we are, so we cannot say the successor is not us.
+		// Keep the older, permissive answer: over-protecting leaks blobs,
+		// guessing the other way deletes them.
+		gc.logger.Warn("GC: own hold DID unknown, treating any successor label as pointing at us",
 			"holdDID", holdDID, "successor", captain.Successor)
 		return true, true
 	}
 
-	// The hold answered and declares no successor. This is the one negative we
-	// are entitled to cache.
-	return false, true
+	if captain.Successor != ourHoldDID {
+		gc.logger.Info("GC: hold has a successor, but it is not us",
+			"holdDID", holdDID, "successor", captain.Successor, "ourHoldDID", ourHoldDID)
+		return false, true
+	}
+
+	gc.logger.Info("GC: discovered predecessor hold (successor points at us)",
+		"holdDID", holdDID, "successor", captain.Successor)
+	return true, true
+}
+
+// ourHoldDID reports this hold's own DID, or "" when it cannot be determined.
+func (gc *GarbageCollector) ourHoldDID() string {
+	if gc.pds == nil {
+		return ""
+	}
+	return gc.pds.DID()
 }
 
 // deleteOrphanedRecords removes layer records whose manifests no longer exist

pkg/hold/gc/predecessor_test.go

@@ -111,6 +111,106 @@ func TestCheckPredecessorAt_InconclusiveOnTransportFailure(t *testing.T) {
 	}
 }
 
+// TestCheckPredecessorAt_DefinitivePositive covers the successful path, which
+// had no test either: a hold that answers and names us as its successor is a
+// predecessor, definitively, and its manifests keep referencing our blobs.
+//
+// The three answers a reachable hold can give are one decision, so they are
+// tested as one table. This is also where GC was brought into line with
+// scan_broadcaster.go:1473, which has always required the successor to name
+// this hold rather than merely to exist.
+func TestCheckPredecessorAt_SuccessorMustNameUs(t *testing.T) {
+	// newRegressionGC gives the collector a real PDS, which is the only way it
+	// learns its own DID. A GC that cannot answer "who am I" takes a different
+	// branch, covered by TestCheckPredecessorAt_UnknownOwnDIDStaysPermissive.
+	gc, holdPDS, ctx := newRegressionGC(t)
+	ourDID := holdPDS.DID()
+	if ourDID == "" {
+		t.Fatal("test hold has no DID: the comparison under test would be skipped entirely")
+	}
+
+	tests := []struct {
+		name              string
+		successor         string
+		wantIsPredecessor bool
+	}{
+		{
+			name:              "successor names us",
+			successor:         ourDID,
+			wantIsPredecessor: true,
+		},
+		{
+			name:              "successor names a third hold",
+			successor:         "did:web:somewhere-else.example.com",
+			wantIsPredecessor: false,
+		},
+		{
+			name:              "no successor at all",
+			successor:         "",
+			wantIsPredecessor: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				// The probe must ask for the captain record of the hold being checked.
+				if got := r.URL.Query().Get("collection"); got != atproto.CaptainCollection {
+					t.Errorf("collection = %q, want %q", got, atproto.CaptainCollection)
+				}
+				if got := r.URL.Query().Get("rkey"); got != "self" {
+					t.Errorf("rkey = %q, want self", got)
+				}
+				w.Header().Set("Content-Type", "application/json")
+				fmt.Fprint(w, captainResponse(t, tt.successor))
+			}))
+			defer srv.Close()
+
+			isPredecessor, definitive := gc.checkPredecessorAt(
+				ctx, "did:web:predecessor.example.com", srv.URL)
+
+			if isPredecessor != tt.wantIsPredecessor {
+				t.Errorf("isPredecessor = %v, want %v", isPredecessor, tt.wantIsPredecessor)
+			}
+			// Every answer here came from a hold that replied, so all three are
+			// cacheable. Only an unanswered probe is inconclusive.
+			if !definitive {
+				t.Error("definitive = false for a hold that answered: the probe is repeated per manifest")
+			}
+		})
+	}
+}
+
+// TestCheckPredecessorAt_UnknownOwnDIDStaysPermissive covers the branch that
+// keeps the successor comparison from becoming a deletion bug. If this hold
+// cannot say who it is, it cannot conclude that a successor is not itself, so
+// it keeps the permissive pre-reconciliation answer. Over-protecting leaks
+// blobs; guessing the other way deletes them.
+func TestCheckPredecessorAt_UnknownOwnDIDStaysPermissive(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, captainResponse(t, "did:web:somewhere-else.example.com"))
+	}))
+	defer srv.Close()
+
+	// No PDS wired, so ourHoldDID() is "".
+	gc := &GarbageCollector{logger: newTestLogger()}
+	if gc.ourHoldDID() != "" {
+		t.Fatal("test GC knows its DID: it would take the comparison branch instead")
+	}
+
+	isPredecessor, definitive := gc.checkPredecessorAt(
+		context.Background(), "did:web:predecessor.example.com", srv.URL)
+
+	if !isPredecessor {
+		t.Error("isPredecessor = false with our own DID unknown: a hold we cannot rule out " +
+			"is treated as unrelated and its blobs become deletable")
+	}
+	if !definitive {
+		t.Error("definitive = false: the hold answered, and our own identity does not change mid-run")
+	}
+}
+
 // TestAnalyzeRecordsClearsPredecessorUnresolved pins the reset documented on
 // the field itself. predecessorUnresolved exists so one unreachable hold costs
 // a single timeout per run rather than one per manifest. If the clear is ever