appview: guard the tag-listing paging that gates a shared-record delete

The delete path decides whether an io.atcr.manifest record is still wanted by
enumerating the DID's tag records. That enumeration is load-bearing in a way a
tag listing usually is not: records for every one of a DID's repositories share
one collection, so a single page is a per-account budget rather than a per-repo
one. Past it a live tag falls off the end, the digest reads as unreferenced,
and the shared record is deleted out from under a repository nobody touched —
along with its layers on the hold.

Two cases, both mutation-verified:

  * the tag that keeps the digest alive sits on page 3. Stopping after the
    first page deletes the record. cleanupUntaggedManifest has carried this
    hazard in a comment since 1b91768 with nothing asserting it.
  * the listing never terminates. Concluding "unreferenced" from an incomplete
    read is the dangerous answer, so this must error rather than guess; making
    it return what it found so far deletes the record.

Both assert on whether a deleteRecord for the manifest collection was issued,
not on the returned error, because the error is incidental and the deletion is
the thing that cannot be undone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SeaUS5AFPX9gqCahoLRMRh

pkg/appview/storage/manifest_store_test.go

@@ -1084,3 +1084,103 @@ func TestManifestStore_Put_RateLimitBecomesErrcode(t *testing.T) {
 		t.Errorf("expected carrier to have a Retry-After duration, got %v", got)
 	}
 }
+
+// tagPageServer serves io.atcr.tag listRecords in pages. The record matching
+// matchDigest is placed on the LAST page, which is the case that matters: a
+// paging bug concludes "untagged" from an incomplete read and deletes a record
+// another repository is still using.
+func tagPageServer(t *testing.T, pages int, matchDigest string, manifestDeletes *int) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case atproto.RepoListRecords:
+			cursor := r.URL.Query().Get("cursor")
+			page := 0
+			if cursor != "" {
+				_, _ = fmt.Sscanf(cursor, "p%d", &page)
+			}
+			last := page >= pages-1
+
+			var recs []string
+			if last && matchDigest != "" {
+				recs = append(recs, fmt.Sprintf(
+					`{"uri":"at://did:plc:test123/io.atcr.tag/otherapp_v1","cid":"bafy","value":{"$type":"io.atcr.tag","repository":"otherapp","tag":"v1","manifest":"at://did:plc:test123/io.atcr.manifest/%s"}}`,
+					strings.TrimPrefix(matchDigest, "sha256:")))
+			} else {
+				// Filler that points somewhere else entirely.
+				recs = append(recs, `{"uri":"at://did:plc:test123/io.atcr.tag/filler","cid":"bafy","value":{"$type":"io.atcr.tag","repository":"filler","tag":"x","manifest":"at://did:plc:test123/io.atcr.manifest/deadbeef"}}`)
+			}
+
+			next := ""
+			if !last {
+				next = fmt.Sprintf(`,"cursor":"p%d"`, page+1)
+			}
+			w.WriteHeader(http.StatusOK)
+			_, _ = fmt.Fprintf(w, `{"records":[%s]%s}`, strings.Join(recs, ","), next)
+
+		case atproto.RepoDeleteRecord:
+			body, _ := io.ReadAll(r.Body)
+			if strings.Contains(string(body), atproto.ManifestCollection) {
+				*manifestDeletes++
+			}
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte(`{"commit":{"cid":"bafytest","rev":"12345"}}`))
+
+		default:
+			w.WriteHeader(http.StatusOK)
+		}
+	}))
+}
+
+// TestManifestStore_Delete_FindsTagOnLaterPage is the paging guard.
+//
+// io.atcr.tag records for every one of a DID's repositories share one
+// collection, so a single page is a per-account budget, not a per-repo one.
+// Past it, a live tag falls off the end, the digest reads as unreferenced, and
+// the shared manifest record is deleted out from under a repository nobody
+// touched — along with its layers on the hold.
+func TestManifestStore_Delete_FindsTagOnLaterPage(t *testing.T) {
+	const dgst = "sha256:abc123"
+	manifestDeletes := 0
+
+	server := tagPageServer(t, 3, dgst, &manifestDeletes)
+	defer server.Close()
+
+	client := atproto.NewClient(server.URL, "did:plc:test123", "token")
+	ctx := mockRegistryContext(client, "myapp", "did:web:hold.example.com", "did:plc:test123", "test.handle", nil)
+	store := NewManifestStore(ctx, nil)
+
+	if err := store.Delete(context.Background(), dgst); err != nil {
+		t.Fatalf("Delete() error = %v, want nil", err)
+	}
+	if manifestDeletes != 0 {
+		t.Errorf("Delete() removed the shared record %d time(s); the tag keeping it alive was on page 3, "+
+			"so the tag listing stopped short", manifestDeletes)
+	}
+}
+
+// TestManifestStore_Delete_PageBudgetFailsClosed covers the other end: a PDS
+// that never stops paging. Guessing past the budget is the dangerous answer, so
+// Delete must error rather than conclude the digest is unreferenced.
+func TestManifestStore_Delete_PageBudgetFailsClosed(t *testing.T) {
+	const dgst = "sha256:abc123"
+	manifestDeletes := 0
+
+	// More pages than the budget, and no record ever matches — so the only
+	// thing stopping a delete is the refusal to decide.
+	server := tagPageServer(t, 1_000_000, "", &manifestDeletes)
+	defer server.Close()
+
+	client := atproto.NewClient(server.URL, "did:plc:test123", "token")
+	ctx := mockRegistryContext(client, "myapp", "did:web:hold.example.com", "did:plc:test123", "test.handle", nil)
+	store := NewManifestStore(ctx, nil)
+
+	err := store.Delete(context.Background(), dgst)
+	if err == nil {
+		t.Error("Delete() returned nil after exhausting the page budget; it must fail closed rather than " +
+			"assume the digest is unreferenced")
+	}
+	if manifestDeletes != 0 {
+		t.Errorf("Delete() removed the shared record %d time(s) despite an incomplete tag listing", manifestDeletes)
+	}
+}