hold/gc: stop the aux-record sweep deleting live records

Follow-up to f12d5e0. Extending the orphan sweep to io.atcr.hold.scan and
io.atcr.hold.image.config brought those collections under a delete path built
for layer records, and the two are not interchangeable. Layer rkeys are
generated per write, so a remembered rkey is a stable handle on one record.
Scan and image-config rkeys are atproto.ScanRecordKey(manifestDigest) — the bare
digest hex, with no DID — which is both reused across re-pushes and shared
between users who push identical images. "Delete rkey K" therefore stopped
meaning "delete the record I judged", and the sweep could destroy live data two
ways.

Both were confirmed by probe before being fixed, and each guard below has a
regression test that fails when that guard alone is reverted.

Shared records: two users pushing the same image collapse onto ONE record whose
body names only the last writer. When that user deleted their manifest the
record looked orphaned, and deleting it stripped the vulnerability scan and the
layer history/env/entrypoint from every co-owner's still-live image.

  - knownDigests keeps a record while any successfully fetched user still holds
    a manifest at that digest, not just the one the record names.

  - knownDigests alone is not enough: it is built only from users we reached, so
    a co-owner whose PDS was down this run is indistinguishable from one who
    deleted their image. digestOwners closes that. It maps each digest to every
    DID that pushed it here, derived from the hold's own layer records during
    the walk analyzeRecords already performs, so it costs nothing extra and
    needs no network. If any owner was unreachable, the record stays. This is
    the sweep's existing principle — an unreachable PDS never causes a deletion —
    extended from the record's named user to everyone the record serves.

    Layer records are the right source because they are load-bearing for storage
    accounting and billing, so they exist for anything a user is charged for.

Reused rkeys: a re-push upserts into the same digest-derived slot, so a record
marked orphaned by an earlier scan can be replaced by a LIVE one before the
delete runs. The admin delete button makes this wide — lastPreview is in-memory
for the life of the process, so an open tab keeps a stale scan actionable — and
doRun has the same race across its analyze-to-delete gap.

  - Records now carry the CID they had when judged, and the delete re-reads the
    slot to confirm it still holds that revision. Anything else (rewritten,
    missing, unreadable, or a ref with no recorded CID) is skipped, not deleted.

  - The CID check is record identity, not orphanhood, and identity alone is not
    enough either. A re-push does not necessarily rewrite the SCAN record:
    scan-on-push is tier-gated, and the broadcaster's discovery loop skips any
    manifest that already has a scan record, so its CID survives a re-push
    unchanged. manifestsWithLayerRecords covers that — a push writes layer
    records to this hold, so their presence proves the manifest is back
    regardless of what the scan concluded. Local, no PDS round trip. A manifest
    whose layer records are themselves orphaned but not yet collected also lands
    in the set, which only defers its aux record one cycle; one more round of a
    leak beats deleting a live user's records.

  - maxPreviewAgeForDelete (30m) refuses a stale preview outright. The
    per-record checks are the real safety net; this stops the admin acting on a
    picture that is hours old. Refusal surfaces through startBackground ->
    lastError -> the polling progress fragment, so it is visible rather than a
    silent no-op.

The four state maps auxRecordOrphaned consults are grouped into auxOrphanState.
Three are same-shaped maps that were trivial to transpose positionally, and
swapping knownDigests with fetchedUsers would have silently widened what the
sweep deletes.

Not addressed, all failure-to-collect rather than data loss: maxPreviewItems
caps the admin button's list while doRun uses the uncapped one, so a hold with
10000+ orphaned layer records surfaces no aux orphans in the preview;
DeleteManifestAuxRecord cannot distinguish "already gone" from "write failed";
and discoverUserDIDs never consults the aux collections, so a record whose owner
has no remaining layer records is kept indefinitely.

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

pkg/hold/gc/aux_records_test.go

@@ -24,6 +24,11 @@ func TestAuxRecordOrphaned(t *testing.T) {
 	knownManifests := map[string]*manifestInfo{
 		knownURI: {URI: knownURI, UserDID: "did:plc:alice"},
 	}
+	// Digests any fetched user still holds. Mirrors knownManifests here; the
+	// two diverge only when users share a digest (see the regression tests).
+	knownDigests := map[string]bool{
+		extractDigestFromManifestURI(knownURI): true,
+	}
 	// alice's PDS answered this run; bob's did not.
 	fetchedUsers := map[string]bool{"did:plc:alice": true}
 
@@ -81,7 +86,11 @@ func TestAuxRecordOrphaned(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			parts, orphan := auxRecordOrphaned(tt.manifestURI, tt.createdAt, knownManifests, fetchedUsers)
+			parts, orphan := auxRecordOrphaned(tt.manifestURI, tt.createdAt, auxOrphanState{
+				knownManifests: knownManifests,
+				knownDigests:   knownDigests,
+				fetchedUsers:   fetchedUsers,
+			})
 			if orphan != tt.wantOrphan {
 				t.Errorf("auxRecordOrphaned(%q) = %v, want %v", tt.manifestURI, orphan, tt.wantOrphan)
 			}
@@ -107,12 +116,12 @@ func TestAuxRecordOrphanedGraceBoundary(t *testing.T) {
 	// A hair past the boundary, to avoid flaking on the clock ticking forward
 	// between constructing the timestamp and evaluating time.Since.
 	justPast := time.Now().Add(-gcRecordGracePeriod - time.Second)
-	if _, orphan := auxRecordOrphaned(uri, justPast, nil, fetchedUsers); !orphan {
+	if _, orphan := auxRecordOrphaned(uri, justPast, auxOrphanState{fetchedUsers: fetchedUsers}); !orphan {
 		t.Error("record just past the grace period should be orphaned")
 	}
 
 	justInside := time.Now().Add(-gcRecordGracePeriod + time.Minute)
-	if _, orphan := auxRecordOrphaned(uri, justInside, nil, fetchedUsers); orphan {
+	if _, orphan := auxRecordOrphaned(uri, justInside, auxOrphanState{fetchedUsers: fetchedUsers}); orphan {
 		t.Error("record just inside the grace period should be kept")
 	}
 }
@@ -291,7 +300,7 @@ func TestDecodeAuxRecordBytesBadTimestampIsKept(t *testing.T) {
 	}
 
 	fetchedUsers := map[string]bool{"did:plc:alice": true}
-	if _, orphan := auxRecordOrphaned(uri, ts, nil, fetchedUsers); orphan {
+	if _, orphan := auxRecordOrphaned(uri, ts, auxOrphanState{fetchedUsers: fetchedUsers}); orphan {
 		t.Error("record with an unparseable timestamp should be kept, not orphaned")
 	}
 }

pkg/hold/gc/config.go

@@ -30,6 +30,14 @@ const (
 	// than this may name a manifest that simply hasn't landed yet. A day is
 	// far longer than any push and still collects on the next nightly run.
 	gcRecordGracePeriod = 24 * time.Hour
+
+	// maxPreviewAgeForDelete bounds how stale a preview may be when the admin
+	// clicks "Delete N Orphaned Records". lastPreview is held in memory for the
+	// life of the process, so an open admin tab keeps that button live and
+	// actionable long after its findings stopped matching reality. The
+	// per-record checks in deleteOrphanedRecords are the real safety net; this
+	// just keeps the admin from acting on a picture that is hours or days old.
+	maxPreviewAgeForDelete = 30 * time.Minute
 )
 
 // Config holds GC configuration

pkg/hold/gc/gc.go

@@ -27,8 +27,14 @@ const maxPreviewItems = 10000
 // names which record type it is — layer records carry a digest, media type,
 // and size; scan and image-config records leave those empty.
 type OrphanedRecordDetail struct {
-	Collection  string `json:"collection"`
-	Rkey        string `json:"rkey"`
+	Collection string `json:"collection"`
+	Rkey       string `json:"rkey"`
+	// CID identifies the exact record revision that was judged orphaned. Scan
+	// and image-config rkeys are derived from the manifest digest and are reused
+	// on re-push, so the rkey alone does not identify what we looked at; the
+	// delete path re-reads this to confirm the slot still holds the same
+	// revision. Empty for layer records, whose rkeys are unique per write.
+	CID         string `json:"cid,omitempty"`
 	Digest      string `json:"digest"`
 	ManifestURI string `json:"manifestUri"`
 	UserDID     string `json:"userDid"`
@@ -36,10 +42,97 @@ type OrphanedRecordDetail struct {
 	Size        int64  `json:"size"`
 }
 
-// orphanRef is the minimal address needed to delete an orphaned record.
+// auxOrphanState is the analysis state auxRecordOrphaned consults. Grouped into
+// a struct rather than passed positionally because three of the four fields are
+// same-shaped maps that would otherwise be trivial to transpose at a call site,
+// and transposing knownDigests with fetchedUsers would silently widen what the
+// sweep deletes.
+type auxOrphanState struct {
+	// knownManifests are the manifests live on their owners' PDSes, by AT-URI.
+	knownManifests map[string]*manifestInfo
+	// knownDigests are the manifest digests any successfully fetched user still
+	// holds, across all users.
+	knownDigests map[string]bool
+	// digestOwners maps a manifest digest to every DID known to have pushed it
+	// to this hold, derived from the hold's own layer records. It is "every
+	// owner we can still see": a user whose layer records were collected in an
+	// earlier run no longer appears, which is self-consistent, since those
+	// records are only collected once that user's manifest is already gone.
+	digestOwners map[string]map[string]bool
+	// fetchedUsers are the DIDs whose PDS answered completely this run.
+	fetchedUsers map[string]bool
+}
+
+// recordDigestOwner notes that the DID in manifestURI owns a manifest at that
+// URI's digest. Called for every layer record the sweep walks, which is what
+// makes the ownership set free to build: layer records are required for storage
+// accounting and billing, so they exist for anything a user is charged for.
+func recordDigestOwner(owners map[string]map[string]bool, manifestURI string) {
+	parts := parseATURI(manifestURI)
+	if parts == nil || parts.Collection != atproto.ManifestCollection {
+		return
+	}
+	digest := "sha256:" + parts.Rkey
+	if owners[digest] == nil {
+		owners[digest] = make(map[string]bool)
+	}
+	owners[digest][parts.DID] = true
+}
+
+// orphanRef is the minimal address needed to delete an orphaned record. CID
+// pins the revision that was judged (see OrphanedRecordDetail.CID); it is empty
+// for layer records, which have unique per-write rkeys and need no such check.
 type orphanRef struct {
 	Collection string
 	Rkey       string
+	CID        string
+	// ManifestURI is the manifest the record belongs to, used at delete time to
+	// check the manifest has not reappeared. Empty for layer records.
+	ManifestURI string
+}
+
+// manifestsWithLayerRecords returns the set of manifest AT-URIs the hold
+// currently holds at least one layer record for.
+//
+// A push writes layer records to this hold, so their presence means the image
+// is here now — regardless of what a previous scan concluded. That makes this a
+// re-check of orphanhood rather than of record identity, and unlike consulting
+// the owner's PDS it is entirely local.
+//
+// A manifest whose layer records are themselves orphaned but not yet collected
+// also lands in this set, which only defers the aux record to the next run.
+// Erring toward one more cycle of a leak is the right trade against deleting a
+// live user's scan and image config.
+func (gc *GarbageCollector) manifestsWithLayerRecords(ctx context.Context) (map[string]bool, error) {
+	recordsIndex := gc.pds.RecordsIndex()
+	if recordsIndex == nil {
+		return nil, fmt.Errorf("records index not available")
+	}
+
+	live := make(map[string]bool)
+	cursor := ""
+	for {
+		records, nextCursor, err := recordsIndex.ListRecords(atproto.LayerCollection, 1000, cursor, true)
+		if err != nil {
+			return nil, fmt.Errorf("list layer records: %w", err)
+		}
+		for _, rec := range records {
+			layer, err := gc.decodeLayerRecord(ctx, rec)
+			if err != nil {
+				// Undecodable: we cannot tell which manifest it belongs to.
+				// Skipping only costs protection, never causes a deletion.
+				gc.logger.Warn("Failed to decode layer record while checking live manifests",
+					"rkey", rec.Rkey, "error", err)
+				continue
+			}
+			live[layer.Manifest] = true
+		}
+		if nextCursor == "" {
+			break
+		}
+		cursor = nextCursor
+	}
+	return live, nil
 }
 
 // OrphanedBlobDetail holds info about a single orphaned blob in S3
@@ -563,11 +656,16 @@ func (gc *GarbageCollector) DeleteOrphanedRecords(ctx context.Context) (*GCResul
 func (gc *GarbageCollector) doDeleteOrphanedRecords(ctx context.Context) (*GCResult, error) {
 	gc.mu.Lock()
 	preview := gc.lastPreview
+	previewAt := gc.lastPreviewAt
 	gc.mu.Unlock()
 
 	if preview == nil {
 		return nil, fmt.Errorf("no preview available — run Scan first")
 	}
+	if age := time.Since(previewAt); age > maxPreviewAgeForDelete {
+		return nil, fmt.Errorf("preview is %s old (limit %s) — run Scan again before deleting",
+			age.Round(time.Minute), maxPreviewAgeForDelete)
+	}
 	if len(preview.OrphanedRecords) == 0 {
 		return &GCResult{}, nil
 	}
@@ -579,7 +677,7 @@ func (gc *GarbageCollector) doDeleteOrphanedRecords(ctx context.Context) (*GCRes
 
 	refs := make([]orphanRef, len(preview.OrphanedRecords))
 	for i, r := range preview.OrphanedRecords {
-		refs[i] = orphanRef{Collection: r.Collection, Rkey: r.Rkey}
+		refs[i] = orphanRef{Collection: r.Collection, Rkey: r.Rkey, CID: r.CID, ManifestURI: r.ManifestURI}
 	}
 
 	gc.logger.Info("Deleting orphaned records", "count", len(refs))
@@ -658,6 +756,15 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult
 
 	// Step 2: Fetch manifests from each user's PDS
 	knownManifests := make(map[string]*manifestInfo)
+	// knownDigests is the set of manifest digests any fetched user still holds,
+	// across ALL users. Scan and image-config records are keyed by digest with no
+	// DID, so they are shared between users pushing identical images and must
+	// survive as long as any one of those users still has the manifest.
+	knownDigests := make(map[string]bool)
+	// digestOwners is built from the hold's own layer records in step 3 below,
+	// giving every DID that pushed a given digest here — the co-owners a shared
+	// scan/image-config record actually serves.
+	digestOwners := make(map[string]map[string]bool)
 	fetchedUsers := make(map[string]bool)
 
 	totalUsers := len(userDIDs)
@@ -726,6 +833,9 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult
 			}
 
 			knownManifests[m.URI] = m
+			if d := extractDigestFromManifestURI(m.URI); d != "" {
+				knownDigests[d] = true
+			}
 
 			// Add all layer digests to referenced set
 			for _, layer := range m.Record.Layers {
@@ -769,6 +879,14 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult
 			pairKey := layer.Manifest + "|" + layer.Digest
 			coveredPairs[pairKey] = true
 
+			// Note the manifest's owner. This MUST stay above every continue
+			// below: an owner missing from the set is an owner the aux sweep
+			// will not wait for, which is how a co-owner's live records get
+			// deleted. The one owner we cannot record is a layer record that
+			// failed to decode above — there is no manifest URI to read — which
+			// is logged as a warning.
+			recordDigestOwner(digestOwners, layer.Manifest)
+
 			// Too young to judge: the manifest may still be in flight. Keep
 			// the record and protect its blob until the record matures.
 			recordTime := tidToTime(rec.Rkey)
@@ -829,8 +947,14 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult
 	// they survive forever when a user deletes a manifest record directly on
 	// their PDS — the purgeManifest XRPC only fires on appview-driven deletes.
 	gc.setProgress("records", "Scanning scan and image config records...", gc.operationType)
+	auxState := auxOrphanState{
+		knownManifests: knownManifests,
+		knownDigests:   knownDigests,
+		digestOwners:   digestOwners,
+		fetchedUsers:   fetchedUsers,
+	}
 	for _, collection := range []string{atproto.ScanCollection, atproto.ImageConfigCollection} {
-		if err := gc.scanAuxRecords(ctx, collection, knownManifests, fetchedUsers, result); err != nil {
+		if err := gc.scanAuxRecords(ctx, collection, auxState, result); err != nil {
 			return nil, fmt.Errorf("scan %s records: %w", collection, err)
 		}
 	}
@@ -870,8 +994,7 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult
 func (gc *GarbageCollector) scanAuxRecords(
 	ctx context.Context,
 	collection string,
-	knownManifests map[string]*manifestInfo,
-	fetchedUsers map[string]bool,
+	state auxOrphanState,
 	result *analysisResult,
 ) error {
 	recordsIndex := gc.pds.RecordsIndex()
@@ -891,27 +1014,30 @@ func (gc *GarbageCollector) scanAuxRecords(
 			scanned++
 			result.totalRecords++
 
-			manifestURI, createdAt, err := gc.decodeAuxRecord(ctx, collection, rec)
+			manifestURI, createdAt, recCID, err := gc.decodeAuxRecord(ctx, collection, rec)
 			if err != nil {
 				gc.logger.Warn("Failed to decode record",
 					"collection", collection, "rkey", rec.Rkey, "error", err)
 				continue
 			}
 
-			parts, isOrphan := auxRecordOrphaned(manifestURI, createdAt, knownManifests, fetchedUsers)
+			parts, isOrphan := auxRecordOrphaned(manifestURI, createdAt, state)
 			if !isOrphan {
 				continue
 			}
 
 			orphaned++
 			result.orphanedRefs = append(result.orphanedRefs, orphanRef{
-				Collection: collection,
-				Rkey:       rec.Rkey,
+				Collection:  collection,
+				Rkey:        rec.Rkey,
+				CID:         recCID,
+				ManifestURI: manifestURI,
 			})
 			if len(result.orphanedDetails) < maxPreviewItems {
 				result.orphanedDetails = append(result.orphanedDetails, OrphanedRecordDetail{
 					Collection:  collection,
 					Rkey:        rec.Rkey,
+					CID:         recCID,
 					ManifestURI: manifestURI,
 					UserDID:     parts.DID,
 				})
@@ -939,20 +1065,40 @@ func (gc *GarbageCollector) scanAuxRecords(
 // window, one whose timestamp couldn't be parsed (zero time), one whose
 // manifest URI is unparseable, and one whose owning PDS was unreachable this
 // run. Only a reachable PDS that demonstrably lacks the manifest orphans it.
-func auxRecordOrphaned(
-	manifestURI string,
-	createdAt time.Time,
-	knownManifests map[string]*manifestInfo,
-	fetchedUsers map[string]bool,
-) (*atURIParts, bool) {
+func auxRecordOrphaned(manifestURI string, createdAt time.Time, state auxOrphanState) (*atURIParts, bool) {
 	if createdAt.IsZero() || time.Since(createdAt) < gcRecordGracePeriod {
 		return nil, false
 	}
-	if _, known := knownManifests[manifestURI]; known {
+	if _, known := state.knownManifests[manifestURI]; known {
 		return nil, false
 	}
 	parts := parseATURI(manifestURI)
-	if parts == nil || !fetchedUsers[parts.DID] {
+	if parts == nil || !state.fetchedUsers[parts.DID] {
+		return nil, false
+	}
+
+	// Scan and image-config records are keyed by manifest digest alone
+	// (atproto.ScanRecordKey), with no DID, so every user who pushes the
+	// identical image shares ONE record — and its body names only whoever wrote
+	// last. The checks above only establish that THAT user no longer has the
+	// manifest, which says nothing about the record's other owners. Deleting on
+	// that basis strips the vulnerability scan and the layer
+	// history/env/entrypoint from every co-owner's still-live image.
+	digest := extractDigestFromManifestURI(manifestURI)
+
+	// An owner whose PDS we could not reach this run cannot be judged, so the
+	// record has to stay. This is the same principle the rest of the sweep
+	// follows — an unreachable PDS never causes a deletion — extended from the
+	// record's named user to everyone the record actually serves.
+	for did := range state.digestOwners[digest] {
+		if !state.fetchedUsers[did] {
+			return nil, false
+		}
+	}
+
+	// Every owner was reachable; keep the record if any of them still holds a
+	// manifest at this digest.
+	if state.knownDigests[digest] {
 		return nil, false
 	}
 	return parts, true
@@ -960,13 +1106,41 @@ func auxRecordOrphaned(
 
 // decodeAuxRecord reads a scan or image-config record from the carstore and
 // returns the manifest AT-URI it belongs to along with its creation time.
-func (gc *GarbageCollector) decodeAuxRecord(ctx context.Context, collection string, rec pds.Record) (string, time.Time, error) {
+// It also returns the record's CID, which the delete path uses to confirm the
+// rkey still holds this exact revision rather than one written since.
+func (gc *GarbageCollector) decodeAuxRecord(ctx context.Context, collection string, rec pds.Record) (string, time.Time, string, error) {
 	recordPath := rec.Collection + "/" + rec.Rkey
-	_, recBytes, err := gc.pds.GetRecordBytes(ctx, recordPath)
+	recCID, recBytes, err := gc.pds.GetRecordBytes(ctx, recordPath)
 	if err != nil {
-		return "", time.Time{}, fmt.Errorf("get record bytes: %w", err)
+		return "", time.Time{}, "", fmt.Errorf("get record bytes: %w", err)
 	}
-	return decodeAuxRecordBytes(collection, *recBytes)
+	uri, ts, err := decodeAuxRecordBytes(collection, *recBytes)
+	return uri, ts, recCID.String(), err
+}
+
+// auxRecordUnchanged reports whether collection/rkey still holds the exact
+// revision that was judged orphaned.
+//
+// This catches a re-push rewriting the slot: image-config records are upserted
+// unconditionally on every push (pkg/hold/oci/xrpc.go), so a re-push always
+// changes their CID. A missing record, an unreadable one, or a ref carrying no
+// recorded CID all report false, so the delete is skipped rather than performed
+// on something we did not examine.
+//
+// It is NOT sufficient on its own. A re-push does not necessarily rewrite the
+// SCAN record — scan-on-push is tier-gated, and the discovery loop skips any
+// manifest that already has a scan record (pkg/hold/pds/scan_broadcaster.go) —
+// so a scan record's CID survives a re-push unchanged. Record identity is not
+// orphanhood; see manifestsWithLayerRecords for the check that covers it.
+func (gc *GarbageCollector) auxRecordUnchanged(ctx context.Context, collection, rkey, wantCID string) bool {
+	if wantCID == "" {
+		return false
+	}
+	gotCID, _, err := gc.pds.GetRecordBytes(ctx, collection+"/"+rkey)
+	if err != nil {
+		return false
+	}
+	return gotCID.String() == wantCID
 }
 
 // decodeAuxRecordBytes decodes raw CBOR for one of the auxiliary manifest
@@ -1694,6 +1868,22 @@ func (gc *GarbageCollector) checkPredecessor(ctx context.Context, holdDID string
 
 // deleteOrphanedRecords removes layer records whose manifests no longer exist
 func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, refs []orphanRef, result *GCResult) error {
+	// Manifests the hold currently holds layer records for. Built once, only if
+	// there is an aux record to delete, and used to confirm at delete time that
+	// the manifest has not come back since it was judged orphaned. Purely local
+	// — no PDS round trip — because a push writes layer records here.
+	var liveManifests map[string]bool
+	for _, ref := range refs {
+		if ref.Collection != "" && ref.Collection != atproto.LayerCollection {
+			var err error
+			if liveManifests, err = gc.manifestsWithLayerRecords(ctx); err != nil {
+				return fmt.Errorf("checking which manifests still have layer records: %w", err)
+			}
+			break
+		}
+	}
+
+	skipped := 0
 	for _, ref := range refs {
 		// An empty collection means the ref came from a preview taken before
 		// the sweep covered scan and image-config records; those were all
@@ -1709,6 +1899,25 @@ func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, refs []or
 				continue
 			}
 		} else {
+			// Confirm the slot still holds the revision we judged. Without this,
+			// a re-push between the scan and the delete would have us destroy a
+			// live record that merely inherited the same digest-derived rkey.
+			if !gc.auxRecordUnchanged(ctx, collection, ref.Rkey, ref.CID) {
+				gc.logger.Info("Skipping record: changed or gone since it was scanned",
+					"collection", collection, "rkey", ref.Rkey, "scannedCID", ref.CID)
+				skipped++
+				continue
+			}
+			// The manifest has layer records again, so it is back on this hold
+			// and the record is in use. Covers the case the CID check cannot: a
+			// re-push does not rewrite the scan record, so its CID still matches
+			// even though the image is live again.
+			if ref.ManifestURI != "" && liveManifests[ref.ManifestURI] {
+				gc.logger.Info("Skipping record: manifest has layer records again",
+					"collection", collection, "rkey", ref.Rkey, "manifest", ref.ManifestURI)
+				skipped++
+				continue
+			}
 			deleted, err := gc.pds.DeleteManifestAuxRecord(ctx, collection, ref.Rkey)
 			if err != nil {
 				gc.logger.Error("Failed to delete record",
@@ -1728,7 +1937,8 @@ func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, refs []or
 
 	gc.logger.Info("Phase 2 complete",
 		"orphaned", len(refs),
-		"deleted", result.RecordsDeleted)
+		"deleted", result.RecordsDeleted,
+		"skippedChangedSinceScan", skipped)
 
 	return nil
 }

pkg/hold/gc/orphan_regression_test.go

@@ -0,0 +1,416 @@
+package gc
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"atcr.io/pkg/atproto"
+	"atcr.io/pkg/hold/pds"
+)
+
+// Regression tests for the two ways the aux-record sweep can delete LIVE user
+// data. Both stem from the same property: unlike layer records, whose rkeys are
+// generated per write, scan and image-config records live at a deterministic
+// rkey derived from the manifest digest alone (atproto.ScanRecordKey). That rkey
+// is reused across re-pushes and shared across users, so "delete rkey K" does
+// not mean "delete the record I judged to be an orphan".
+
+const regressionDigest = "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f"
+
+// TestAuxRecordOrphaned_DigestSharedAcrossUsers covers two users who pushed the
+// identical image to one hold. Content-addressed dedup means both manifests have
+// the same digest, so both users' scan and image-config records collapse onto the
+// single rkey hex(digest), and the record body names whichever user wrote last.
+//
+// When that last writer deletes their manifest, the record looks orphaned — but
+// the OTHER user's manifest is still live and still needs it. Deleting it strips
+// a live image of its vulnerability scan and its layer history/env/entrypoint.
+func TestAuxRecordOrphaned_DigestSharedAcrossUsers(t *testing.T) {
+	aliceURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+	bobURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest)
+
+	// Premise check: the two users genuinely collide on one record key. If this
+	// ever stops being true (e.g. the rkey gains a DID component) this whole
+	// failure mode is gone and the test should be revisited rather than patched.
+	if aliceURI == bobURI {
+		t.Fatal("test setup: URIs should differ by DID")
+	}
+	if got, want := atproto.ScanRecordKey(regressionDigest), atproto.ScanRecordKey(regressionDigest); got != want {
+		t.Fatalf("rkey not deterministic: %q vs %q", got, want)
+	}
+
+	// Alice's manifest is live and was fetched successfully. Bob's is gone.
+	// Both PDSes answered this run, so neither is excused by unreachability.
+	knownManifests := map[string]*manifestInfo{
+		aliceURI: {URI: aliceURI, UserDID: "did:plc:alice"},
+	}
+	fetchedUsers := map[string]bool{
+		"did:plc:alice": true,
+		"did:plc:bob":   true,
+	}
+
+	knownDigests := map[string]bool{
+		extractDigestFromManifestURI(aliceURI): true,
+	}
+	// Both users pushed this digest here, so the hold holds layer records for
+	// both — that is where the co-ownership is visible.
+	digestOwners := map[string]map[string]bool{
+		extractDigestFromManifestURI(aliceURI): {"did:plc:alice": true, "did:plc:bob": true},
+	}
+
+	// The record at hex(digest) names Bob, the last writer.
+	_, isOrphan := auxRecordOrphaned(bobURI, pastGrace, auxOrphanState{
+		knownManifests: knownManifests,
+		knownDigests:   knownDigests,
+		digestOwners:   digestOwners,
+		fetchedUsers:   fetchedUsers,
+	})
+
+	if isOrphan {
+		t.Fatalf("record at rkey %s judged orphaned because Bob's manifest is gone, "+
+			"but Alice still has a live manifest at the same digest — deleting it "+
+			"destroys the scan and image config for Alice's live image",
+			atproto.ScanRecordKey(regressionDigest))
+	}
+}
+
+// TestAuxRecordOrphaned_DigestUniqueToUser is the companion: when the ONLY user
+// holding that digest deletes their manifest, the record really is garbage and
+// must still be collected. This guards the fix for the test above from being
+// implemented as a blanket "never delete", which would silently reinstate the
+// leak the sweep exists to fix.
+func TestAuxRecordOrphaned_DigestUniqueToUser(t *testing.T) {
+	bobURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest)
+	otherURI := atproto.BuildManifestURI("did:plc:alice", "sha256:0000000000000000000000000000000000000000000000000000000000000000")
+
+	// Alice is live but at a DIFFERENT digest, so she does not protect this record.
+	knownManifests := map[string]*manifestInfo{
+		otherURI: {URI: otherURI, UserDID: "did:plc:alice"},
+	}
+	fetchedUsers := map[string]bool{
+		"did:plc:alice": true,
+		"did:plc:bob":   true,
+	}
+
+	// Alice's live manifest is at a different digest, so it protects nothing here.
+	knownDigests := map[string]bool{
+		extractDigestFromManifestURI(otherURI): true,
+	}
+	// Only Bob ever pushed this digest, so nobody else's records depend on it.
+	digestOwners := map[string]map[string]bool{
+		extractDigestFromManifestURI(bobURI): {"did:plc:bob": true},
+	}
+
+	parts, isOrphan := auxRecordOrphaned(bobURI, pastGrace, auxOrphanState{
+		knownManifests: knownManifests,
+		knownDigests:   knownDigests,
+		digestOwners:   digestOwners,
+		fetchedUsers:   fetchedUsers,
+	})
+	if !isOrphan {
+		t.Fatal("record whose only referencing manifest was deleted should be collected")
+	}
+	if parts == nil || parts.DID != "did:plc:bob" {
+		t.Fatalf("expected owning DID did:plc:bob, got %+v", parts)
+	}
+}
+
+// newRegressionGC builds a GarbageCollector backed by a real embedded PDS.
+// doDeleteOrphanedRecords only touches gc.pds, gc.lastPreview and gc.logger, so
+// no S3 service is needed.
+func newRegressionGC(t *testing.T) (*GarbageCollector, *pds.HoldPDS, context.Context) {
+	t.Helper()
+
+	ctx := context.Background()
+	tmp := t.TempDir()
+
+	holdPDS, err := pds.NewHoldPDS(
+		ctx,
+		"did:web:hold.example.com",
+		"https://hold.example.com",
+		"https://atcr.io",
+		filepath.Join(tmp, "pds.db"),
+		filepath.Join(tmp, "signing-key"),
+		false,
+	)
+	if err != nil {
+		t.Fatalf("NewHoldPDS: %v", err)
+	}
+	t.Cleanup(func() { holdPDS.Close() })
+
+	if err := holdPDS.Bootstrap(ctx, nil, pds.BootstrapConfig{OwnerDID: "did:plc:owner", Public: true}); err != nil {
+		t.Fatalf("Bootstrap: %v", err)
+	}
+
+	// Wire records indexing before any test record is written: the delete path
+	// consults the index to see which manifests still have layer records.
+	holdPDS.RepomgrRef().SetEventHandler(holdPDS.CreateRecordsIndexEventHandler(nil), true)
+	if err := holdPDS.BackfillRecordsIndex(ctx); err != nil {
+		t.Fatalf("BackfillRecordsIndex: %v", err)
+	}
+
+	gc := &GarbageCollector{
+		pds:    holdPDS,
+		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
+	}
+	return gc, holdPDS, ctx
+}
+
+// seedOrphanedConfig writes an image-config record and returns its rkey, path
+// and CID — the CID being what the preview records so the delete path can tell
+// this exact revision from whatever occupies the slot later.
+func seedOrphanedConfig(t *testing.T, holdPDS *pds.HoldPDS, ctx context.Context, manifestURI, configJSON string) (rkey, path, recCID string) {
+	t.Helper()
+	_, c, err := holdPDS.CreateImageConfigRecord(ctx,
+		atproto.NewImageConfigRecord(manifestURI, configJSON), regressionDigest)
+	if err != nil {
+		t.Fatalf("writing image config record: %v", err)
+	}
+	rkey = atproto.ScanRecordKey(regressionDigest)
+	return rkey, atproto.ImageConfigCollection + "/" + rkey, c.String()
+}
+
+func previewOf(rkey, recCID, manifestURI string, at time.Time) (*GCPreview, time.Time) {
+	return &GCPreview{
+		OrphanedRecords: []OrphanedRecordDetail{{
+			Collection:  atproto.ImageConfigCollection,
+			Rkey:        rkey,
+			CID:         recCID,
+			ManifestURI: manifestURI,
+			UserDID:     "did:plc:alice",
+		}},
+	}, at
+}
+
+// TestDeleteOrphanedRecords_StalePreviewIsRefused covers the admin "Delete N
+// Orphaned Records" button, which the GC page re-renders from the last preview
+// on every load — so without a limit a scan stays actionable for days while its
+// findings drift from reality.
+func TestDeleteOrphanedRecords_StalePreviewIsRefused(t *testing.T) {
+	gc, holdPDS, ctx := newRegressionGC(t)
+	manifestURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+
+	rkey, recordPath, recCID := seedOrphanedConfig(t, holdPDS, ctx, manifestURI, `{"orphan":true}`)
+	gc.lastPreview, gc.lastPreviewAt = previewOf(rkey, recCID, manifestURI,
+		time.Now().Add(-maxPreviewAgeForDelete-time.Hour))
+
+	if _, err := gc.doDeleteOrphanedRecords(ctx); err == nil {
+		t.Fatal("a preview older than maxPreviewAgeForDelete should be refused, not replayed")
+	}
+	if _, _, err := holdPDS.GetRecordBytes(ctx, recordPath); err != nil {
+		t.Errorf("refused delete should not have touched the record: %v", err)
+	}
+}
+
+// TestDeleteOrphanedRecords_RewrittenRecordIsNotDeleted is the core regression.
+// Scan and image-config records live at a deterministic rkey derived from the
+// manifest digest, and CreateImageConfigRecord upserts, so re-pushing an image
+// puts a NEW, LIVE record in the exact slot an earlier scan marked for deletion.
+// Layer records were immune because their rkeys are unique per write, so a stale
+// rkey simply resolved to nothing.
+//
+// The preview here is fresh, so the age limit does not apply — this pins the CID
+// check on its own.
+func TestDeleteOrphanedRecords_RewrittenRecordIsNotDeleted(t *testing.T) {
+	gc, holdPDS, ctx := newRegressionGC(t)
+	manifestURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+
+	// Scan time: the manifest is gone, so this record is a genuine orphan.
+	rkey, recordPath, scannedCID := seedOrphanedConfig(t, holdPDS, ctx, manifestURI, `{"stale":true}`)
+	gc.lastPreview, gc.lastPreviewAt = previewOf(rkey, scannedCID, manifestURI, time.Now())
+
+	// Before the admin clicks: Alice re-pushes. Same digest, so the upsert lands
+	// on the same rkey and the record now belongs to a live manifest.
+	_, _, rewrittenCID := seedOrphanedConfig(t, holdPDS, ctx, manifestURI, `{"live":true}`)
+	if rewrittenCID == scannedCID {
+		t.Fatal("test setup: re-push should have produced a different record revision")
+	}
+
+	result, err := gc.doDeleteOrphanedRecords(ctx)
+	if err != nil {
+		t.Fatalf("doDeleteOrphanedRecords: %v", err)
+	}
+	if result.RecordsDeleted != 0 {
+		t.Errorf("RecordsDeleted = %d, want 0 (the slot was rewritten since the scan)", result.RecordsDeleted)
+	}
+	if _, _, err := holdPDS.GetRecordBytes(ctx, recordPath); err != nil {
+		t.Fatalf("deleted the LIVE image config at %s after a re-push: %v", recordPath, err)
+	}
+}
+
+// TestDeleteOrphanedRecords_UnchangedRecordIsDeleted is the companion guard: a
+// fresh preview whose records are untouched must still collect them. Without
+// this, both fixes above could be "refuse everything" and the sweep would
+// silently stop doing its job.
+func TestDeleteOrphanedRecords_UnchangedRecordIsDeleted(t *testing.T) {
+	gc, holdPDS, ctx := newRegressionGC(t)
+	manifestURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+
+	rkey, recordPath, recCID := seedOrphanedConfig(t, holdPDS, ctx, manifestURI, `{"orphan":true}`)
+	gc.lastPreview, gc.lastPreviewAt = previewOf(rkey, recCID, manifestURI, time.Now())
+
+	result, err := gc.doDeleteOrphanedRecords(ctx)
+	if err != nil {
+		t.Fatalf("doDeleteOrphanedRecords: %v", err)
+	}
+	if result.RecordsDeleted != 1 {
+		t.Errorf("RecordsDeleted = %d, want 1", result.RecordsDeleted)
+	}
+	if _, _, err := holdPDS.GetRecordBytes(ctx, recordPath); err == nil {
+		t.Errorf("orphaned record at %s survived an unchanged, fresh preview delete", recordPath)
+	}
+}
+
+// TestAuxRecordOrphaned_UnreachableCoOwnerIsKept covers the hole that the
+// digest-sharing check alone leaves open.
+//
+// knownDigests is built only from users whose PDS answered, so a co-owner who
+// was unreachable this run contributes nothing to it — she looks identical to a
+// user who deleted her manifest. The reachability check earlier in
+// auxRecordOrphaned does not help either: it tests the DID named in the RECORD
+// (the last writer), not the other users the shared record serves.
+//
+// The sweep's stated principle is that an unreachable PDS never causes a
+// deletion. digestOwners is what extends that from the record's named user to
+// everyone it actually serves.
+func TestAuxRecordOrphaned_UnreachableCoOwnerIsKept(t *testing.T) {
+	aliceURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+	bobURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest)
+	digest := extractDigestFromManifestURI(bobURI)
+
+	// Bob answered and no longer has the manifest. Alice's PDS was down, so she
+	// is absent from both fetchedUsers and knownDigests — indistinguishable,
+	// without digestOwners, from a user who deleted her image.
+	state := auxOrphanState{
+		knownManifests: map[string]*manifestInfo{},
+		knownDigests:   map[string]bool{},
+		digestOwners: map[string]map[string]bool{
+			digest: {"did:plc:alice": true, "did:plc:bob": true},
+		},
+		fetchedUsers: map[string]bool{"did:plc:bob": true},
+	}
+
+	if _, orphan := auxRecordOrphaned(bobURI, pastGrace, state); orphan {
+		t.Fatalf("record at digest %s judged orphaned while co-owner did:plc:alice "+
+			"was unreachable this run; an unreachable PDS must never cause a deletion "+
+			"(alice's manifest URI would be %s)", digest, aliceURI)
+	}
+}
+
+// TestAuxRecordOrphaned_AllOwnersReachableAndGoneIsCollected is the companion:
+// once every owner has been reached and none of them still holds the manifest,
+// the record really is garbage and must be collected. Without this the fix could
+// degrade into "keep anything with more than one owner", reinstating the leak.
+func TestAuxRecordOrphaned_AllOwnersReachableAndGoneIsCollected(t *testing.T) {
+	bobURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest)
+	digest := extractDigestFromManifestURI(bobURI)
+
+	state := auxOrphanState{
+		knownManifests: map[string]*manifestInfo{},
+		knownDigests:   map[string]bool{},
+		digestOwners: map[string]map[string]bool{
+			digest: {"did:plc:alice": true, "did:plc:bob": true},
+		},
+		fetchedUsers: map[string]bool{"did:plc:alice": true, "did:plc:bob": true},
+	}
+
+	parts, orphan := auxRecordOrphaned(bobURI, pastGrace, state)
+	if !orphan {
+		t.Fatal("record whose every owner was reached and no longer holds the manifest should be collected")
+	}
+	if parts == nil || parts.DID != "did:plc:bob" {
+		t.Fatalf("expected owning DID did:plc:bob, got %+v", parts)
+	}
+}
+
+// TestRecordDigestOwner covers the derivation itself: ownership comes from the
+// hold's own layer records, which exist for anything a user is billed for.
+func TestRecordDigestOwner(t *testing.T) {
+	owners := make(map[string]map[string]bool)
+
+	aliceURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+	bobURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest)
+	recordDigestOwner(owners, aliceURI)
+	recordDigestOwner(owners, bobURI)
+	recordDigestOwner(owners, aliceURI) // repeat layers of the same manifest
+
+	digest := extractDigestFromManifestURI(aliceURI)
+	if got := len(owners[digest]); got != 2 {
+		t.Errorf("owners for %s = %d, want 2 (alice and bob, deduped)", digest, got)
+	}
+	if !owners[digest]["did:plc:alice"] || !owners[digest]["did:plc:bob"] {
+		t.Errorf("missing an owner: %+v", owners[digest])
+	}
+
+	// Junk must not create phantom owners that would keep records alive forever.
+	for _, bad := range []string{"", "not-an-at-uri", "at://did:plc:alice/io.atcr.hold.layer/xyz"} {
+		before := len(owners)
+		recordDigestOwner(owners, bad)
+		if len(owners) != before {
+			t.Errorf("recordDigestOwner(%q) added an entry; non-manifest URIs must be ignored", bad)
+		}
+	}
+}
+
+// TestDeleteOrphanedRecords_RepushedScanRecordIsNotDeleted covers the case the
+// CID check cannot see.
+//
+// A re-push always rewrites the image-config record (CreateImageConfigRecord is
+// called unconditionally on push), so its CID changes and the identity check
+// catches it. A re-push does NOT necessarily rewrite the SCAN record:
+// scan-on-push is tier-gated, and the scan broadcaster's discovery loop skips
+// any manifest that already has a scan record. So the scan record's CID still
+// matches the preview even though the image is live again.
+//
+// What does change is that the push writes layer records back to this hold, and
+// deleteOrphanedRecords consults those.
+func TestDeleteOrphanedRecords_RepushedScanRecordIsNotDeleted(t *testing.T) {
+	gc, holdPDS, ctx := newRegressionGC(t)
+	manifestURI := atproto.BuildManifestURI("did:plc:alice", regressionDigest)
+	rkey := atproto.ScanRecordKey(regressionDigest)
+	recordPath := atproto.ScanCollection + "/" + rkey
+
+	// Scan time: the manifest was gone, so this scan record was a real orphan.
+	scanRec := &atproto.ScanRecord{
+		Type:      atproto.ScanCollection,
+		Manifest:  manifestURI,
+		UserDID:   "did:plc:alice",
+		ScannedAt: time.Now().Add(-72 * time.Hour).Format(time.RFC3339),
+	}
+	_, scanCID, err := holdPDS.CreateScanRecord(ctx, scanRec)
+	if err != nil {
+		t.Fatalf("seeding scan record: %v", err)
+	}
+	gc.lastPreview = &GCPreview{
+		OrphanedRecords: []OrphanedRecordDetail{{
+			Collection:  atproto.ScanCollection,
+			Rkey:        rkey,
+			CID:         scanCID.String(),
+			ManifestURI: manifestURI,
+			UserDID:     "did:plc:alice",
+		}},
+	}
+	gc.lastPreviewAt = time.Now()
+
+	// Alice re-pushes: layer records come back, the scan record is untouched.
+	if _, _, err := holdPDS.CreateLayerRecord(ctx, atproto.NewLayerRecord(
+		"sha256:layer-a", 1024, "application/vnd.oci.image.layer.v1.tar+gzip",
+		"did:plc:alice", manifestURI)); err != nil {
+		t.Fatalf("re-push layer record: %v", err)
+	}
+
+	result, err := gc.doDeleteOrphanedRecords(ctx)
+	if err != nil {
+		t.Fatalf("doDeleteOrphanedRecords: %v", err)
+	}
+	if result.RecordsDeleted != 0 {
+		t.Errorf("RecordsDeleted = %d, want 0 (the manifest is back on this hold)", result.RecordsDeleted)
+	}
+	if _, _, err := holdPDS.GetRecordBytes(ctx, recordPath); err != nil {
+		t.Fatalf("deleted the scan record for a re-pushed, live manifest at %s: %v", recordPath, err)
+	}
+}