diff --git a/pkg/hold/gc/aux_records_test.go b/pkg/hold/gc/aux_records_test.go index 9d57648..1269cfa 100644 --- a/pkg/hold/gc/aux_records_test.go +++ b/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") } } diff --git a/pkg/hold/gc/config.go b/pkg/hold/gc/config.go index 7afe7f9..c848e41 100644 --- a/pkg/hold/gc/config.go +++ b/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 diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index 7404234..8d86ee7 100644 --- a/pkg/hold/gc/gc.go +++ b/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 } diff --git a/pkg/hold/gc/orphan_regression_test.go b/pkg/hold/gc/orphan_regression_test.go new file mode 100644 index 0000000..3840fd7 --- /dev/null +++ b/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) + } +}