diff --git a/pkg/hold/admin/templates/partials/gc_preview.html b/pkg/hold/admin/templates/partials/gc_preview.html index 864b815..1225f52 100644 --- a/pkg/hold/admin/templates/partials/gc_preview.html +++ b/pkg/hold/admin/templates/partials/gc_preview.html @@ -40,22 +40,24 @@ {{if .Preview.OrphanedRecords}}
- +
- {{ icon "file-x" "size-4" }} Orphaned Layer Records ({{len .Preview.OrphanedRecords}}) + {{ icon "file-x" "size-4" }} Orphaned Records ({{len .Preview.OrphanedRecords}})
- + - - - - + + + + + + @@ -65,6 +67,7 @@ {{range .Preview.OrphanedRecords}} + @@ -171,7 +174,7 @@ hx-post="/admin/api/gc/delete-records" hx-target="#gc-results" hx-swap="innerHTML" - hx-confirm="Delete {{len .Preview.OrphanedRecords}} orphaned layer records?"> + hx-confirm="Delete {{len .Preview.OrphanedRecords}} orphaned records?"> {{ icon "file-x" "size-4" }} Delete {{len .Preview.OrphanedRecords}} Orphaned Records diff --git a/pkg/hold/gc/aux_records_test.go b/pkg/hold/gc/aux_records_test.go new file mode 100644 index 0000000..9d57648 --- /dev/null +++ b/pkg/hold/gc/aux_records_test.go @@ -0,0 +1,297 @@ +package gc + +import ( + "bytes" + "testing" + "time" + + "atcr.io/pkg/atproto" +) + +// pastGrace is comfortably older than gcRecordGracePeriod; inGrace is newer. +var ( + pastGrace = time.Now().Add(-gcRecordGracePeriod - 24*time.Hour) + inGrace = time.Now().Add(-1 * time.Minute) +) + +func TestAuxRecordOrphaned(t *testing.T) { + const ( + knownURI = "at://did:plc:alice/io.atcr.manifest/abc123" + deletedURI = "at://did:plc:alice/io.atcr.manifest/def456" + offlineURI = "at://did:plc:bob/io.atcr.manifest/ghi789" + ) + + knownManifests := map[string]*manifestInfo{ + knownURI: {URI: knownURI, UserDID: "did:plc:alice"}, + } + // alice's PDS answered this run; bob's did not. + fetchedUsers := map[string]bool{"did:plc:alice": true} + + tests := []struct { + name string + manifestURI string + createdAt time.Time + wantOrphan bool + wantDID string + }{ + { + name: "manifest deleted on reachable PDS is orphaned", + manifestURI: deletedURI, + createdAt: pastGrace, + wantOrphan: true, + wantDID: "did:plc:alice", + }, + { + name: "manifest still present is kept", + manifestURI: knownURI, + createdAt: pastGrace, + wantOrphan: false, + }, + { + name: "record inside grace window is kept", + manifestURI: deletedURI, + createdAt: inGrace, + wantOrphan: false, + }, + { + name: "unreachable PDS is kept even though manifest is unknown", + manifestURI: offlineURI, + createdAt: pastGrace, + wantOrphan: false, + }, + { + name: "zero timestamp is kept", + manifestURI: deletedURI, + createdAt: time.Time{}, + wantOrphan: false, + }, + { + name: "unparseable manifest URI is kept", + manifestURI: "not-an-at-uri", + createdAt: pastGrace, + wantOrphan: false, + }, + { + name: "empty manifest URI is kept", + manifestURI: "", + createdAt: pastGrace, + wantOrphan: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + parts, orphan := auxRecordOrphaned(tt.manifestURI, tt.createdAt, knownManifests, fetchedUsers) + if orphan != tt.wantOrphan { + t.Errorf("auxRecordOrphaned(%q) = %v, want %v", tt.manifestURI, orphan, tt.wantOrphan) + } + if !orphan { + return + } + if parts == nil { + t.Fatal("expected parsed AT-URI parts for an orphaned record, got nil") + } + if parts.DID != tt.wantDID { + t.Errorf("parts.DID = %q, want %q", parts.DID, tt.wantDID) + } + }) + } +} + +// A record exactly at the grace boundary must be collected, not kept — the +// boundary test is `< gcRecordGracePeriod`, so equal-or-older is eligible. +func TestAuxRecordOrphanedGraceBoundary(t *testing.T) { + const uri = "at://did:plc:alice/io.atcr.manifest/gone" + fetchedUsers := map[string]bool{"did:plc:alice": true} + + // 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 { + 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 { + t.Error("record just inside the grace period should be kept") + } +} + +// Blobs age on their own clock, independent of the records that name them. +// Their window is deliberately longer than the record window, so a blob whose +// records were already collected still survives until its own age is reached. +func TestBlobPastGrace(t *testing.T) { + tests := []struct { + name string + lastModified time.Time + want bool + }{ + { + name: "blob older than the blob grace period is collectable", + lastModified: time.Now().Add(-gcBlobGracePeriod - time.Hour), + want: true, + }, + { + name: "blob inside the blob grace period is kept", + lastModified: time.Now().Add(-gcBlobGracePeriod + time.Hour), + want: false, + }, + { + name: "freshly uploaded blob is kept", + lastModified: time.Now(), + want: false, + }, + { + name: "unknown modification time is kept", + lastModified: time.Time{}, + want: false, + }, + { + name: "blob past the record window but not the blob window is kept", + // The case that motivates the split: records for this blob are + // already eligible for collection, the bytes are not. + lastModified: time.Now().Add(-gcRecordGracePeriod - time.Hour), + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := blobPastGrace(tt.lastModified); got != tt.want { + t.Errorf("blobPastGrace(%v) = %v, want %v", tt.lastModified, got, tt.want) + } + }) + } +} + +// The split only holds if blobs outlive records. If these are ever reordered, +// a blob could be deleted while a live record still names it. +func TestBlobGraceOutlivesRecordGrace(t *testing.T) { + if gcBlobGracePeriod <= gcRecordGracePeriod { + t.Errorf("gcBlobGracePeriod (%v) must exceed gcRecordGracePeriod (%v): blobs must outlive the records naming them", + gcBlobGracePeriod, gcRecordGracePeriod) + } +} + +func TestDecodeAuxRecordBytes(t *testing.T) { + const ( + manifestURI = "at://did:plc:alice/io.atcr.manifest/abc123" + scannedAt = "2026-03-04T05:06:07Z" + createdAt = "2026-01-02T03:04:05Z" + ) + + scan := &atproto.ScanRecord{ + Type: atproto.ScanCollection, + Manifest: manifestURI, + UserDID: "did:plc:alice", + ScannedAt: scannedAt, + } + var scanBuf bytes.Buffer + if err := scan.MarshalCBOR(&scanBuf); err != nil { + t.Fatalf("marshal scan record: %v", err) + } + + cfg := &atproto.ImageConfigRecord{ + Type: atproto.ImageConfigCollection, + Manifest: manifestURI, + ConfigJSON: `{"architecture":"amd64"}`, + CreatedAt: createdAt, + } + var cfgBuf bytes.Buffer + if err := cfg.MarshalCBOR(&cfgBuf); err != nil { + t.Fatalf("marshal image config record: %v", err) + } + + tests := []struct { + name string + collection string + data []byte + wantURI string + wantTime string // RFC3339, or "" for the zero time + wantErr bool + }{ + { + name: "scan record", + collection: atproto.ScanCollection, + data: scanBuf.Bytes(), + wantURI: manifestURI, + wantTime: scannedAt, + }, + { + name: "image config record", + collection: atproto.ImageConfigCollection, + data: cfgBuf.Bytes(), + wantURI: manifestURI, + wantTime: createdAt, + }, + { + name: "unsupported collection is rejected", + collection: atproto.LayerCollection, + data: scanBuf.Bytes(), + wantErr: true, + }, + { + name: "malformed CBOR is an error", + collection: atproto.ScanCollection, + data: []byte("not cbor at all"), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + uri, ts, err := decodeAuxRecordBytes(tt.collection, tt.data) + if tt.wantErr { + if err == nil { + t.Fatal("expected an error, got nil") + } + return + } + if err != nil { + t.Fatalf("decodeAuxRecordBytes: %v", err) + } + if uri != tt.wantURI { + t.Errorf("manifest URI = %q, want %q", uri, tt.wantURI) + } + want, parseErr := time.Parse(time.RFC3339, tt.wantTime) + if parseErr != nil { + t.Fatalf("bad want time in test case: %v", parseErr) + } + if !ts.Equal(want) { + t.Errorf("timestamp = %v, want %v", ts, want) + } + }) + } +} + +// A record whose timestamp doesn't parse must decode without error but yield +// the zero time, so auxRecordOrphaned keeps it rather than collecting it. +func TestDecodeAuxRecordBytesBadTimestampIsKept(t *testing.T) { + const manifestURI = "at://did:plc:alice/io.atcr.manifest/abc123" + + scan := &atproto.ScanRecord{ + Type: atproto.ScanCollection, + Manifest: manifestURI, + ScannedAt: "yesterday-ish", + } + var buf bytes.Buffer + if err := scan.MarshalCBOR(&buf); err != nil { + t.Fatalf("marshal scan record: %v", err) + } + + uri, ts, err := decodeAuxRecordBytes(atproto.ScanCollection, buf.Bytes()) + if err != nil { + t.Fatalf("decodeAuxRecordBytes: %v", err) + } + if uri != manifestURI { + t.Errorf("manifest URI = %q, want %q", uri, manifestURI) + } + if !ts.IsZero() { + t.Errorf("timestamp = %v, want the zero time", ts) + } + + fetchedUsers := map[string]bool{"did:plc:alice": true} + if _, orphan := auxRecordOrphaned(uri, ts, nil, 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 af5a489..7afe7f9 100644 --- a/pkg/hold/gc/config.go +++ b/pkg/hold/gc/config.go @@ -1,6 +1,11 @@ // Package gc implements garbage collection for the hold service. -// It periodically cleans up orphaned blobs from S3 storage based on -// layer records in the hold's embedded PDS. +// +// It collects two things on separate clocks: records in the hold's embedded +// PDS (layer, scan, and image config) whose manifest no longer exists in the +// owning user's PDS, and blobs in S3 that no live manifest references. Records +// go first and blobs follow later, so a user who deletes a manifest record +// directly on their PDS stops being billed for it well before the bytes are +// actually reclaimed. package gc import "time" @@ -10,9 +15,21 @@ const ( // gcInterval is how often GC runs (nightly) gcInterval = 24 * time.Hour - // gcGracePeriod is how old a layer record must be before it's considered for GC. - // Records created in the last 7 days are skipped (GDPR/CCPA compliant). - gcGracePeriod = 7 * 24 * time.Hour + // gcBlobGracePeriod is how old a blob must be, by its own S3 modification + // time, before GC will delete it. Blob pruning is best-effort: reclaiming + // bytes a week late costs storage, reclaiming them early can destroy + // content a client is still pushing or a takedown may yet reverse. + gcBlobGracePeriod = 7 * 24 * time.Hour + + // gcRecordGracePeriod is how old a record must be before GC treats a + // missing manifest as an intentional deletion rather than a race. + // + // Records are metadata, not content, so they don't need the blob window. + // What they do need is to outlast a push: blobs and layer records are + // written before the manifest reaches the user's PDS, so a record younger + // 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 ) // Config holds GC configuration diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index 841bb00..7404234 100644 --- a/pkg/hold/gc/gc.go +++ b/pkg/hold/gc/gc.go @@ -23,8 +23,11 @@ import ( // maxPreviewItems caps per-category detail slices to prevent memory/HTML bloat const maxPreviewItems = 10000 -// OrphanedRecordDetail holds info about a single orphaned layer record +// OrphanedRecordDetail holds info about a single orphaned record. Collection +// 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"` Digest string `json:"digest"` ManifestURI string `json:"manifestUri"` @@ -33,6 +36,12 @@ type OrphanedRecordDetail struct { Size int64 `json:"size"` } +// orphanRef is the minimal address needed to delete an orphaned record. +type orphanRef struct { + Collection string + Rkey string +} + // OrphanedBlobDetail holds info about a single orphaned blob in S3 type OrphanedBlobDetail struct { Digest string `json:"digest"` @@ -175,7 +184,7 @@ type manifestInfo struct { // analysisResult holds intermediate data from record analysis, shared between Run and Preview type analysisResult struct { referenced map[string]bool - orphanedRkeys []string // rkeys for deletion in Run + orphanedRefs []orphanRef // collection+rkey for deletion in Run orphanedDetails []OrphanedRecordDetail // details for display in Preview missingDetails []MissingRecordDetail // details for creation in Run / display in Preview usersChecked int64 @@ -404,13 +413,13 @@ func (gc *GarbageCollector) doRun(ctx context.Context) (*GCResult, error) { return nil, fmt.Errorf("phase 1 (analyze records) failed: %w", err) } - result.OrphanedRecords = int64(len(analysis.orphanedRkeys)) + result.OrphanedRecords = int64(len(analysis.orphanedRefs)) result.UsersChecked = analysis.usersChecked result.ManifestsChecked = analysis.manifestsChecked gc.logger.Info("Phase 1 complete", "referenced", len(analysis.referenced), - "orphanedRecords", len(analysis.orphanedRkeys), + "orphanedRecords", len(analysis.orphanedRefs), "missingRecords", len(analysis.missingDetails)) // Reconcile: create missing layer records @@ -421,7 +430,7 @@ func (gc *GarbageCollector) doRun(ctx context.Context) (*GCResult, error) { // Phase 2: Delete orphaned layer records gc.setProgress("deleting", "Deleting orphaned records...", "run") - if err := gc.deleteOrphanedRecords(ctx, analysis.orphanedRkeys, result); err != nil { + if err := gc.deleteOrphanedRecords(ctx, analysis.orphanedRefs, result); err != nil { gc.logger.Error("Phase 2 (delete orphaned records) failed", "error", err) // Continue to phase 3 - we can still clean up blobs } @@ -568,13 +577,13 @@ func (gc *GarbageCollector) doDeleteOrphanedRecords(ctx context.Context) (*GCRes OrphanedRecords: int64(len(preview.OrphanedRecords)), } - rkeys := make([]string, len(preview.OrphanedRecords)) + refs := make([]orphanRef, len(preview.OrphanedRecords)) for i, r := range preview.OrphanedRecords { - rkeys[i] = r.Rkey + refs[i] = orphanRef{Collection: r.Collection, Rkey: r.Rkey} } - gc.logger.Info("Deleting orphaned records", "count", len(rkeys)) - if err := gc.deleteOrphanedRecords(ctx, rkeys, result); err != nil { + gc.logger.Info("Deleting orphaned records", "count", len(refs)) + if err := gc.deleteOrphanedRecords(ctx, refs, result); err != nil { return nil, fmt.Errorf("delete orphaned records: %w", err) } result.Duration = time.Since(start) @@ -760,9 +769,10 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult pairKey := layer.Manifest + "|" + layer.Digest coveredPairs[pairKey] = true - // Grace period: skip recent records + // 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) - if time.Since(recordTime) < gcGracePeriod { + if time.Since(recordTime) < gcRecordGracePeriod { result.referenced[layer.Digest] = true continue } @@ -778,9 +788,13 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult result.referenced[layer.Digest] = true } else { // User's PDS was reachable but manifest not found — orphaned - result.orphanedRkeys = append(result.orphanedRkeys, rec.Rkey) + result.orphanedRefs = append(result.orphanedRefs, orphanRef{ + Collection: atproto.LayerCollection, + Rkey: rec.Rkey, + }) if len(result.orphanedDetails) < maxPreviewItems { result.orphanedDetails = append(result.orphanedDetails, OrphanedRecordDetail{ + Collection: atproto.LayerCollection, Rkey: rec.Rkey, Digest: layer.Digest, ManifestURI: layer.Manifest, @@ -809,6 +823,18 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult gc.logger.Info("Scanned layer records", "total", result.totalRecords, "coveredPairs", len(coveredPairs)) + // Step 3b: Scan scan and image-config records. These are keyed by manifest + // digest rather than TID and hold no blob references of their own, so the + // only question is whether their manifest still exists. Without this sweep + // 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) + for _, collection := range []string{atproto.ScanCollection, atproto.ImageConfigCollection} { + if err := gc.scanAuxRecords(ctx, collection, knownManifests, fetchedUsers, result); err != nil { + return nil, fmt.Errorf("scan %s records: %w", collection, err) + } + } + // Step 4: Identify missing layer records (uncovered manifest+layer pairs) for _, m := range knownManifests { for _, layer := range m.Record.Layers { @@ -831,6 +857,162 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult return result, nil } +// scanAuxRecords walks one auxiliary manifest collection (scan or image +// config) and appends orphans to result. A record is orphaned when its +// manifest is absent from the owning user's PDS and that PDS was reachable — +// the same test the layer sweep applies, so an unreachable PDS never causes a +// deletion. +// +// Grace is taken from the record body (scannedAt / createdAt) rather than the +// rkey, because these collections use deterministic digest rkeys, not TIDs. +// An unparseable or absent timestamp is treated as in-grace, so a malformed +// record is kept rather than collected. +func (gc *GarbageCollector) scanAuxRecords( + ctx context.Context, + collection string, + knownManifests map[string]*manifestInfo, + fetchedUsers map[string]bool, + result *analysisResult, +) error { + recordsIndex := gc.pds.RecordsIndex() + if recordsIndex == nil { + return fmt.Errorf("records index not available") + } + + cursor := "" + scanned, orphaned := 0, 0 + for { + records, nextCursor, err := recordsIndex.ListRecords(collection, 1000, cursor, true) + if err != nil { + return fmt.Errorf("list records: %w", err) + } + + for _, rec := range records { + scanned++ + result.totalRecords++ + + manifestURI, createdAt, 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) + if !isOrphan { + continue + } + + orphaned++ + result.orphanedRefs = append(result.orphanedRefs, orphanRef{ + Collection: collection, + Rkey: rec.Rkey, + }) + if len(result.orphanedDetails) < maxPreviewItems { + result.orphanedDetails = append(result.orphanedDetails, OrphanedRecordDetail{ + Collection: collection, + Rkey: rec.Rkey, + ManifestURI: manifestURI, + UserDID: parts.DID, + }) + } + gc.logger.Debug("Found orphaned record", + "collection", collection, "rkey", rec.Rkey, "manifest", manifestURI) + } + + if nextCursor == "" { + break + } + cursor = nextCursor + } + + gc.logger.Info("Scanned auxiliary manifest records", + "collection", collection, "scanned", scanned, "orphaned", orphaned) + return nil +} + +// auxRecordOrphaned decides whether a scan or image-config record should be +// collected, given the manifest it names and its creation time. It returns the +// parsed AT-URI alongside the verdict so the caller can record the owning DID. +// +// Every uncertain case resolves to "keep": a record still inside the grace +// 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) { + if createdAt.IsZero() || time.Since(createdAt) < gcRecordGracePeriod { + return nil, false + } + if _, known := knownManifests[manifestURI]; known { + return nil, false + } + parts := parseATURI(manifestURI) + if parts == nil || !fetchedUsers[parts.DID] { + return nil, false + } + return parts, true +} + +// 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) { + recordPath := rec.Collection + "/" + rec.Rkey + _, recBytes, err := gc.pds.GetRecordBytes(ctx, recordPath) + if err != nil { + return "", time.Time{}, fmt.Errorf("get record bytes: %w", err) + } + return decodeAuxRecordBytes(collection, *recBytes) +} + +// decodeAuxRecordBytes decodes raw CBOR for one of the auxiliary manifest +// collections. An unparseable timestamp yields the zero time rather than an +// error, which auxRecordOrphaned treats as in-grace, so a record with a +// malformed date is kept instead of collected. +func decodeAuxRecordBytes(collection string, data []byte) (string, time.Time, error) { + var manifestURI, timestamp string + switch collection { + case atproto.ScanCollection: + var scan atproto.ScanRecord + if err := scan.UnmarshalCBOR(bytes.NewReader(data)); err != nil { + return "", time.Time{}, fmt.Errorf("unmarshal CBOR: %w", err) + } + manifestURI, timestamp = scan.Manifest, scan.ScannedAt + case atproto.ImageConfigCollection: + var cfg atproto.ImageConfigRecord + if err := cfg.UnmarshalCBOR(bytes.NewReader(data)); err != nil { + return "", time.Time{}, fmt.Errorf("unmarshal CBOR: %w", err) + } + manifestURI, timestamp = cfg.Manifest, cfg.CreatedAt + default: + return "", time.Time{}, fmt.Errorf("unsupported collection %q", collection) + } + + parsed, err := time.Parse(time.RFC3339, timestamp) + if err != nil { + return manifestURI, time.Time{}, nil + } + return manifestURI, parsed, nil +} + +// blobPastGrace reports whether a blob is old enough to delete, based on its +// own S3 modification time rather than on any record that references it. +// +// Records now age out faster than blobs, so a blob can outlive every record +// naming it. Its age is the only thing left to judge it by. A listing that +// reports no modification time yields the zero time, which is treated as +// "unknown age" and keeps the blob. +func blobPastGrace(lastModified time.Time) bool { + if lastModified.IsZero() { + return false + } + return time.Since(lastModified) >= gcBlobGracePeriod +} + // scanOrphanedBlobDetails walks S3 and returns details of unreferenced blobs. // Read-only — no deletions. Returns orphaned blob details and total blob count. func (gc *GarbageCollector) scanOrphanedBlobDetails(ctx context.Context, referenced map[string]bool) ([]OrphanedBlobDetail, int, error) { @@ -838,7 +1020,7 @@ func (gc *GarbageCollector) scanOrphanedBlobDetails(ctx context.Context, referen totalBlobs := 0 blobsPath := "/docker/registry/v2/blobs" - err := gc.s3.WalkBlobs(ctx, blobsPath, func(key string, size int64) error { + err := gc.s3.WalkBlobs(ctx, blobsPath, func(key string, size int64, lastModified time.Time) error { if !strings.HasSuffix(key, "/data") { return nil } @@ -850,7 +1032,7 @@ func (gc *GarbageCollector) scanOrphanedBlobDetails(ctx context.Context, referen totalBlobs++ - if !referenced[digest] { + if !referenced[digest] && blobPastGrace(lastModified) { if len(orphaned) < maxPreviewItems { orphaned = append(orphaned, OrphanedBlobDetail{ Digest: digest, @@ -1511,18 +1693,41 @@ 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, orphanedRkeys []string, result *GCResult) error { - for _, rkey := range orphanedRkeys { - if err := gc.pds.DeleteLayerRecord(ctx, rkey); err != nil { - gc.logger.Error("Failed to delete layer record", "rkey", rkey, "error", err) - continue +func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, refs []orphanRef, result *GCResult) error { + 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 + // layer records. + collection := ref.Collection + if collection == "" { + collection = atproto.LayerCollection } + + if collection == atproto.LayerCollection { + if err := gc.pds.DeleteLayerRecord(ctx, ref.Rkey); err != nil { + gc.logger.Error("Failed to delete layer record", "rkey", ref.Rkey, "error", err) + continue + } + } else { + deleted, err := gc.pds.DeleteManifestAuxRecord(ctx, collection, ref.Rkey) + if err != nil { + gc.logger.Error("Failed to delete record", + "collection", collection, "rkey", ref.Rkey, "error", err) + continue + } + if !deleted { + // Already gone (e.g. a concurrent purge) — not an error, but + // don't count it as work this run did. + continue + } + } + result.RecordsDeleted++ - gc.logger.Debug("Deleted orphaned layer record", "rkey", rkey) + gc.logger.Debug("Deleted orphaned record", "collection", collection, "rkey", ref.Rkey) } gc.logger.Info("Phase 2 complete", - "orphaned", len(orphanedRkeys), + "orphaned", len(refs), "deleted", result.RecordsDeleted) return nil @@ -1532,7 +1737,7 @@ func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, orphanedR func (gc *GarbageCollector) deleteOrphanedBlobs(ctx context.Context, referenced map[string]bool, result *GCResult) error { blobsPath := "/docker/registry/v2/blobs" - err := gc.s3.WalkBlobs(ctx, blobsPath, func(key string, size int64) error { + err := gc.s3.WalkBlobs(ctx, blobsPath, func(key string, size int64, lastModified time.Time) error { // Only process data files if !strings.HasSuffix(key, "/data") { return nil @@ -1549,6 +1754,13 @@ func (gc *GarbageCollector) deleteOrphanedBlobs(ctx context.Context, referenced return nil } + // Unreferenced, but young enough that we'd rather keep paying for it + // than risk deleting content still being pushed. Records for this + // blob may already be gone; blob pruning runs on its own clock. + if !blobPastGrace(lastModified) { + return nil + } + result.OrphanedBlobs++ if err := gc.s3.Delete(ctx, key); err != nil { diff --git a/pkg/hold/pds/purge.go b/pkg/hold/pds/purge.go index 087e4ed..44e539b 100644 --- a/pkg/hold/pds/purge.go +++ b/pkg/hold/pds/purge.go @@ -72,6 +72,23 @@ func (p *HoldPDS) PurgeManifestRecords(ctx context.Context, manifestURI string) return res, nil } +// DeleteManifestAuxRecord deletes a single scan or image-config record by its +// deterministic digest rkey. Restricted to those two collections so callers +// can't reach captain, crew, or layer records through it — layer records have +// their own typed deletion path (DeleteLayerRecord). +// +// Idempotent: a missing record reports false with no error. Used by the GC's +// orphaned-record sweep, which addresses these records directly rather than +// going through PurgeManifestRecords (it already knows they're orphaned). +func (p *HoldPDS) DeleteManifestAuxRecord(ctx context.Context, collection, rkey string) (bool, error) { + switch collection { + case atproto.ScanCollection, atproto.ImageConfigCollection: + default: + return false, fmt.Errorf("collection %q is not an auxiliary manifest record", collection) + } + return p.tryDeleteRecord(ctx, collection, rkey), nil +} + // PurgeUserManifests purges every manifest's records for a given DID. Used by // user-level takedowns (URI = at://) where the labeler has not enumerated // individual manifest URIs. diff --git a/pkg/hold/pds/purge_test.go b/pkg/hold/pds/purge_test.go index 4b4197c..ace011e 100644 --- a/pkg/hold/pds/purge_test.go +++ b/pkg/hold/pds/purge_test.go @@ -5,6 +5,7 @@ import ( "testing" "atcr.io/pkg/atproto" + "github.com/ipfs/go-cid" ) func TestPurgeManifestRecordsRemovesAll(t *testing.T) { @@ -125,6 +126,78 @@ func TestPurgeUserManifestsCollectsAcrossManifests(t *testing.T) { } } +func TestDeleteManifestAuxRecord(t *testing.T) { + pds := setupTestPDSWithIndex(t, "did:plc:owner") + ctx := sharedCtx + + const manifestDigest = "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f" + manifestURI := atproto.BuildManifestURI("did:plc:alice", manifestDigest) + rkey := atproto.ScanRecordKey(manifestDigest) + + scanRec := &atproto.ScanRecord{ + Type: atproto.ScanCollection, + Manifest: manifestURI, + ScannedAt: "2026-05-02T00:00:00Z", + } + if _, _, _, err := pds.repomgr.UpsertRecord(ctx, pds.uid, atproto.ScanCollection, rkey, scanRec); err != nil { + t.Fatalf("create scan record: %v", err) + } + + // A layer record for the same manifest must survive — this entry point + // addresses only the deterministic-rkey auxiliary records. + mustCreateLayer(t, pds, manifestURI, "sha256:layer-a", 1024) + + deleted, err := pds.DeleteManifestAuxRecord(ctx, atproto.ScanCollection, rkey) + if err != nil { + t.Fatalf("DeleteManifestAuxRecord: %v", err) + } + if !deleted { + t.Error("deleted = false, want true for an existing scan record") + } + if _, _, err := pds.repomgr.GetRecord(ctx, pds.uid, atproto.ScanCollection, rkey, cid.Undef); err == nil { + t.Error("scan record still present after deletion") + } + + rkeys, err := pds.listLayerRkeysForManifest(ctx, manifestURI) + if err != nil { + t.Fatalf("list layers: %v", err) + } + if len(rkeys) != 1 { + t.Errorf("layer rkeys = %d, want 1 (aux delete must not touch layer records)", len(rkeys)) + } + + // Deleting again is not an error, it just reports nothing was removed. + deleted, err = pds.DeleteManifestAuxRecord(ctx, atproto.ScanCollection, rkey) + if err != nil { + t.Fatalf("second DeleteManifestAuxRecord: %v", err) + } + if deleted { + t.Error("deleted = true on second call, want false") + } +} + +func TestDeleteManifestAuxRecordRejectsOtherCollections(t *testing.T) { + pds := setupTestPDSWithIndex(t, "did:plc:owner") + + // Layer, captain, and crew records must not be reachable through this + // entry point — a GC bug shouldn't be able to delete the hold's identity. + for _, collection := range []string{ + atproto.LayerCollection, + atproto.CaptainCollection, + atproto.CrewCollection, + } { + t.Run(collection, func(t *testing.T) { + deleted, err := pds.DeleteManifestAuxRecord(sharedCtx, collection, "self") + if err == nil { + t.Errorf("DeleteManifestAuxRecord(%q) returned nil error, want rejection", collection) + } + if deleted { + t.Errorf("DeleteManifestAuxRecord(%q) reported a deletion", collection) + } + }) + } +} + // mustCreateLayer is a tiny helper for purge tests — keeps the table-driven // TestPurge body focused on the assertions. func mustCreateLayer(t *testing.T, pds *HoldPDS, manifestURI, digest string, size int64) { diff --git a/pkg/s3/types.go b/pkg/s3/types.go index 39809e3..7e53219 100644 --- a/pkg/s3/types.go +++ b/pkg/s3/types.go @@ -334,7 +334,11 @@ func (s *S3Service) Delete(ctx context.Context, blobPath string) error { // WalkBlobs paginates ListObjectsV2 under prefix and calls fn for each object. // Keys passed to fn have the PathPrefix stripped (same format as BlobPath output). -func (s *S3Service) WalkBlobs(ctx context.Context, prefix string, fn func(key string, size int64) error) error { +// +// lastModified is the object's S3 modification time, which callers use to age +// blobs independently of any record that references them. It is the zero time +// if the listing didn't report one. +func (s *S3Service) WalkBlobs(ctx context.Context, prefix string, fn func(key string, size int64, lastModified time.Time) error) error { s3Prefix := s.s3Key(prefix) if !strings.HasSuffix(s3Prefix, "/") { s3Prefix += "/" @@ -367,7 +371,11 @@ func (s *S3Service) WalkBlobs(ctx context.Context, prefix string, fn func(key st if obj.Size != nil { size = *obj.Size } - if err := fn(key, size); err != nil { + var lastModified time.Time + if obj.LastModified != nil { + lastModified = *obj.LastModified + } + if err := fn(key, size, lastModified); err != nil { return err } }
Orphaned layer recordsOrphaned layer, scan, and image config records
Collection RKey Digest Manifest
{{.Collection}} {{.Rkey}} {{truncate .Digest 24}} {{truncate .ManifestURI 50}}