diff --git a/pkg/hold/gc/aux_grace_test.go b/pkg/hold/gc/aux_grace_test.go new file mode 100644 index 0000000..fd9dda2 --- /dev/null +++ b/pkg/hold/gc/aux_grace_test.go @@ -0,0 +1,427 @@ +package gc + +import ( + "testing" + "time" + + "atcr.io/pkg/atproto" +) + +// testRescanInterval is a hold configured well below gcRecordGracePeriod. +// Nothing enforces a lower bound on scanner.rescan_interval, so this is a legal +// configuration. +const testRescanInterval = 6 * time.Hour + +// orphanedScanState is the analysis state for a scan record whose only owner +// answered and no longer holds the manifest: a genuine, unambiguous orphan. +func orphanedScanState() (recordKey, manifestURI string, state auxOrphanState) { + manifestURI = atproto.BuildManifestURI("did:plc:bob", regressionDigest) + digest := extractDigestFromManifestURI(manifestURI) + recordKey = auxRecordKey(atproto.ScanCollection, atproto.ScanRecordKey(regressionDigest)) + + return recordKey, manifestURI, auxOrphanState{ + knownManifests: map[string]*manifestInfo{}, + knownDigests: map[string]bool{}, + digestOwners: map[string]map[string]bool{digest: {"did:plc:bob": true}}, + fetchedUsers: map[string]bool{"did:plc:bob": true}, + } +} + +// withClock returns a copy of state wired to one pass of the orphan clock: +// judged at now, reading carried, writing into a fresh observation map. +func withClock(state auxOrphanState, now time.Time, carried map[string]time.Time) (auxOrphanState, map[string]time.Time) { + observed := make(map[string]time.Time) + state.now = now + state.orphanSince = carried + state.orphanObserved = observed + return state, observed +} + +// TestAuxRecordOrphaned_RescannedRecordIsEventuallyCollected pins the grace +// clock to orphanhood rather than to the record body. +// +// io.atcr.hold.scan records carry exactly one timestamp, scannedAt, and the +// stale-scan loop rewrites it on every rescan (pkg/hold/pds/scan_broadcaster.go +// requeues anything older than rescan_interval, and both the success and the +// failure constructors stamp scannedAt with time.Now). A hold whose +// rescan_interval is shorter than gcRecordGracePeriod therefore presents GC +// with a record that is permanently "young", however long its manifest has +// been gone. Reading grace off that timestamp made such a record immortal. +// +// Simulate it: a nightly GC pass, with the stale-scan loop rewriting the record +// between every pair of passes. The manifest is gone throughout and its only +// owner answered every time, so the record must eventually be collected. +func TestAuxRecordOrphaned_RescannedRecordIsEventuallyCollected(t *testing.T) { + recordKey, manifestURI, base := orphanedScanState() + + const passes = 30 // a month of nightly runs + start := time.Now() + carried := map[string]time.Time{} + + for pass := range passes { + now := start.Add(time.Duration(pass) * gcInterval) + + // The record's own scannedAt, restamped within the last rescan + // interval. It is deliberately not passed to auxRecordOrphaned any + // more; it is here to say what the record looks like on this pass. + if scannedAt := now.Add(-testRescanInterval / 2); pass > 0 && scannedAt.Before(start) { + t.Fatalf("test setup: scannedAt %v should track the pass clock", scannedAt) + } + + state, observed := withClock(base, now, carried) + _, orphan := auxRecordOrphaned(recordKey, manifestURI, state) + carried = observed + + if orphan { + if pass == 0 { + t.Fatalf("collected on the first pass; a record must survive at least "+ + "gcRecordGracePeriod (%v) of observed orphanhood", gcRecordGracePeriod) + } + return + } + } + + t.Fatalf("an orphaned scan record rescanned every %v was never collected across %d "+ + "nightly passes (%v of wall time); the grace clock is reading the record body, "+ + "which every rescan resets, instead of how long the record has been orphaned", + testRescanInterval, passes, time.Duration(passes)*gcInterval) +} + +// The clock must not expire early. One pass, however long the process has been +// up, can never collect: grace is a duration of OBSERVED orphanhood, and a +// single observation is a duration of zero. +func TestAuxRecordOrphaned_FirstObservationNeverCollects(t *testing.T) { + recordKey, manifestURI, base := orphanedScanState() + + state, _ := withClock(base, time.Now(), map[string]time.Time{}) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("a slot observed orphaned for the first time must be kept") + } +} + +// A slot that stops looking orphaned loses its accrued age, and starts over if +// it ever looks orphaned again. This is what protects a re-pushed image: the +// pass that sees the manifest back does not observe the slot, commit drops it, +// and any later deletion has to wait out a fresh grace period. +func TestAuxRecordOrphaned_ClockResetsWhenManifestReturns(t *testing.T) { + recordKey, manifestURI, base := orphanedScanState() + start := time.Now() + + // Pass 1: orphaned, clock starts. + state, observed := withClock(base, start, map[string]time.Time{}) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("first observation should not collect") + } + if _, tracked := observed[recordKey]; !tracked { + t.Fatal("first observation should have started the clock for the slot") + } + carried := observed + + // Pass 2, a day later: the user re-pushed, so the manifest is live again. + // The slot must not be observed, and commit's replace semantics mean the + // entry is gone from what the next pass carries. + live := base + live.knownManifests = map[string]*manifestInfo{manifestURI: {URI: manifestURI, UserDID: "did:plc:bob"}} + live.knownDigests = map[string]bool{extractDigestFromManifestURI(manifestURI): true} + state, observed = withClock(live, start.Add(gcInterval), carried) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("a slot whose manifest is live must never be collected") + } + if _, tracked := observed[recordKey]; tracked { + t.Fatal("a live slot must not accrue orphan age") + } + carried = observed + + // Pass 3, a day after that: the manifest is gone again. Because pass 2 + // dropped the entry, this is a first observation, not a day-old one. + state, observed = withClock(base, start.Add(2*gcInterval), carried) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("grace must restart after the slot came back to life, not resume") + } + carried = observed + + // And a day after THAT it is collectable, so the reset is a delay and not + // a permanent reprieve. + state, _ = withClock(base, start.Add(3*gcInterval), carried) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); !orphan { + t.Fatal("a slot orphaned continuously for a full grace period should be collected") + } +} + +// A record whose owner was unreachable this pass is not judged, and must not +// accrue age either. The sweep's standing rule is that an unreachable PDS never +// contributes to a deletion; a clock ticking in the background while a hold is +// down would break that quietly, a run later. +func TestAuxRecordOrphaned_UnreachableOwnerDoesNotAccrueAge(t *testing.T) { + recordKey, manifestURI, base := orphanedScanState() + + offline := base + offline.fetchedUsers = map[string]bool{} + + state, observed := withClock(offline, time.Now(), map[string]time.Time{}) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("a record whose owning PDS was unreachable must be kept") + } + if _, tracked := observed[recordKey]; tracked { + t.Fatal("an unreachable owner's record must not accrue orphan age") + } +} + +// A restart empties the clock, so every surviving orphan starts its grace +// period over. That costs a day of retention, which is the safe direction, and +// it is the tradeoff for the clock being in-memory rather than persisted. +func TestAuxRecordOrphaned_RestartRestartsGrace(t *testing.T) { + recordKey, manifestURI, base := orphanedScanState() + start := time.Now() + + // A pass just before the restart, which would have collected on the next. + state, observed := withClock(base, start, map[string]time.Time{}) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("first observation should not collect") + } + if len(observed) != 1 { + t.Fatalf("expected the slot to be tracked, got %d entries", len(observed)) + } + + // Restart: a fresh process carries nothing, so the pass that would have + // collected instead re-observes from scratch. + state, observed = withClock(base, start.Add(gcInterval), map[string]time.Time{}) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("after a restart the clock must start over, not collect on the first pass") + } + + // And it collects a grace period after the restart, so a restart delays + // collection rather than preventing it. + state, _ = withClock(base, start.Add(2*gcInterval), observed) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); !orphan { + t.Fatal("a restarted clock should still collect a full grace period later") + } +} + +// The image-config sweep changed along with the scan sweep, so this pins that +// the change never collects one EARLIER than the old createdAt rule would have. +// The old clock started when the record was written; the new one starts when GC +// first observed it orphaned, which cannot be before it was written. So for the +// collection whose timestamp did not reset, the general fix is strictly the +// more conservative of the two. +func TestAuxRecordOrphaned_ImageConfigIsNotCollectedEarlier(t *testing.T) { + manifestURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest) + digest := extractDigestFromManifestURI(manifestURI) + recordKey := auxRecordKey(atproto.ImageConfigCollection, atproto.ScanRecordKey(regressionDigest)) + + base := auxOrphanState{ + knownManifests: map[string]*manifestInfo{}, + knownDigests: map[string]bool{}, + digestOwners: map[string]map[string]bool{digest: {"did:plc:bob": true}}, + fetchedUsers: map[string]bool{"did:plc:bob": true}, + } + + // The record was written well before GC first looked at it. Under the old + // rule createdAt alone made it collectable on sight. + createdAt := time.Now().Add(-30 * 24 * time.Hour) + firstPass := time.Now() + if !firstPass.After(createdAt.Add(gcRecordGracePeriod)) { + t.Fatal("test setup: the record should be past grace by its createdAt") + } + + state, observed := withClock(base, firstPass, map[string]time.Time{}) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); orphan { + t.Fatal("an image-config record must not be collected on the pass that first observed it, " + + "however old its createdAt") + } + + state, _ = withClock(base, firstPass.Add(gcRecordGracePeriod), observed) + if _, orphan := auxRecordOrphaned(recordKey, manifestURI, state); !orphan { + t.Fatal("an image-config record orphaned across a full grace period should be collected") + } +} + +// graceElapsed is the whole clock, so its edges are worth pinning directly: +// every uncertain input must start the clock rather than expire it. +func TestGraceElapsed(t *testing.T) { + now := time.Now() + const key = "io.atcr.hold.scan/abc" + + tests := []struct { + name string + since map[string]time.Time + want bool + }{ + { + name: "unseen slot starts the clock", + since: map[string]time.Time{}, + want: false, + }, + { + name: "seen a full grace period ago is elapsed", + since: map[string]time.Time{key: now.Add(-gcRecordGracePeriod)}, + want: true, + }, + { + name: "seen just inside the window is not elapsed", + since: map[string]time.Time{key: now.Add(-gcRecordGracePeriod + time.Minute)}, + want: false, + }, + { + name: "zero time starts the clock rather than expiring it", + since: map[string]time.Time{key: {}}, + want: false, + }, + { + // A clock that jumped backwards would otherwise leave a future + // timestamp behind and expire it as soon as time.Now caught up. + name: "a future observation restarts the clock", + since: map[string]time.Time{key: now.Add(time.Hour)}, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + observed := make(map[string]time.Time) + state := auxOrphanState{now: now, orphanSince: tt.since, orphanObserved: observed} + if got := state.graceElapsed(key); got != tt.want { + t.Errorf("graceElapsed(%q) = %v, want %v", key, got, tt.want) + } + if _, tracked := observed[key]; !tracked { + t.Error("graceElapsed must record the observation it was asked about") + } + }) + } +} + +// A nil orphanObserved map means nothing is carried forward, so no slot ever +// accrues age. That has to be a keep-everything failure, not a panic and not a +// delete-everything one. +func TestGraceElapsedNilObservedIsSafe(t *testing.T) { + state := auxOrphanState{now: time.Now()} + if state.graceElapsed("io.atcr.hold.scan/abc") { + t.Error("with no clock wired up, no record may be judged past grace") + } +} + +// The clock's own carry-forward semantics: commit replaces, so a slot the pass +// did not observe is forgotten rather than left ticking. +func TestAuxOrphanClockCommitPrunes(t *testing.T) { + var clock auxOrphanClock + now := time.Now() + + clock.commit(map[string]time.Time{"a": now, "b": now}) + if got := clock.snapshot(); len(got) != 2 { + t.Fatalf("snapshot after commit = %d entries, want 2", len(got)) + } + + // The next pass observed only "a". + clock.commit(map[string]time.Time{"a": now}) + got := clock.snapshot() + if _, ok := got["b"]; ok { + t.Error("a slot the pass did not observe must be dropped, not carried") + } + if !got["a"].Equal(now) { + t.Errorf("an observed slot must keep its ORIGINAL first-seen time, got %v want %v", got["a"], now) + } + + // snapshot hands out a copy: mutating it must not reach back into the clock. + got["c"] = now + if _, ok := clock.snapshot()["c"]; ok { + t.Error("snapshot must return a copy, not the live map") + } +} + +// The zero-value clock is the one every GarbageCollector starts with, so it has +// to work without a constructor. +func TestAuxOrphanClockZeroValue(t *testing.T) { + var clock auxOrphanClock + if got := clock.snapshot(); len(got) != 0 { + t.Errorf("zero-value clock snapshot = %d entries, want 0", len(got)) + } +} + +// TestScanAuxRecords_RescanDoesNotResetGrace is the same bug at the level the +// sweep actually runs at: a real embedded PDS, a real scan record, and a real +// rewrite of that record between passes. +// +// The unit test above proves the decision function times orphanhood. This +// proves the sweep feeds it a slot key that survives a rescan — key it by CID, +// or by anything the rewrite changes, and the clock silently restarts on every +// rescan exactly as reading scannedAt did. +func TestScanAuxRecords_RescanDoesNotResetGrace(t *testing.T) { + gc, holdPDS, ctx := newRegressionGC(t) + + manifestURI := atproto.BuildManifestURI("did:plc:bob", regressionDigest) + digest := extractDigestFromManifestURI(manifestURI) + rkey := atproto.ScanRecordKey(regressionDigest) + + writeScan := func(scannedAt time.Time) string { + t.Helper() + _, cid, err := holdPDS.CreateScanRecord(ctx, &atproto.ScanRecord{ + Type: atproto.ScanCollection, + Manifest: manifestURI, + UserDID: "did:plc:bob", + ScannedAt: scannedAt.Format(time.RFC3339), + }) + if err != nil { + t.Fatalf("writing scan record: %v", err) + } + return cid.String() + } + + // Bob's manifest is gone from his PDS, and his PDS answered. + newState := func(now time.Time, carried map[string]time.Time) (auxOrphanState, map[string]time.Time) { + observed := make(map[string]time.Time) + return auxOrphanState{ + knownManifests: map[string]*manifestInfo{}, + knownDigests: map[string]bool{}, + digestOwners: map[string]map[string]bool{digest: {"did:plc:bob": true}}, + fetchedUsers: map[string]bool{"did:plc:bob": true}, + now: now, + orphanSince: carried, + orphanObserved: observed, + }, observed + } + + sweep := func(state auxOrphanState) []orphanRef { + t.Helper() + result := &analysisResult{referenced: map[string]bool{}} + if err := gc.scanAuxRecords(ctx, atproto.ScanCollection, state, result); err != nil { + t.Fatalf("scanAuxRecords: %v", err) + } + return result.orphanedRefs + } + + start := time.Now() + + // Pass 1: first sighting. Nothing may be collected yet. + firstCID := writeScan(start.Add(-testRescanInterval / 2)) + state, observed := newState(start, map[string]time.Time{}) + if refs := sweep(state); len(refs) != 0 { + t.Fatalf("pass 1 reported %d orphans; a first sighting must collect nothing", len(refs)) + } + if _, tracked := observed[auxRecordKey(atproto.ScanCollection, rkey)]; !tracked { + t.Fatal("pass 1 should have started the orphan clock for the scan record") + } + carried := observed + + // Between passes the stale-scan loop rescans and rewrites the record: same + // slot, new revision, freshly stamped scannedAt. + secondCID := writeScan(start.Add(gcInterval).Add(-testRescanInterval / 2)) + if secondCID == firstCID { + t.Fatal("test setup: the rescan should have produced a different record revision") + } + + // Pass 2, a grace period later. The rewrite must not have bought the record + // another day. + state, _ = newState(start.Add(gcRecordGracePeriod), carried) + refs := sweep(state) + if len(refs) != 1 { + t.Fatalf("pass 2 reported %d orphans, want 1: a rescan must not reset the grace clock", len(refs)) + } + if refs[0].Rkey != rkey || refs[0].Collection != atproto.ScanCollection { + t.Errorf("orphan ref = %s/%s, want %s/%s", refs[0].Collection, refs[0].Rkey, atproto.ScanCollection, rkey) + } + // The ref must pin the revision the sweep actually looked at, so the delete + // path can still refuse a slot rewritten after this pass. + if refs[0].CID != secondCID { + t.Errorf("orphan ref CID = %q, want the revision this pass read (%q)", refs[0].CID, secondCID) + } +} diff --git a/pkg/hold/gc/aux_records_test.go b/pkg/hold/gc/aux_records_test.go index 1269cfa..e10ba66 100644 --- a/pkg/hold/gc/aux_records_test.go +++ b/pkg/hold/gc/aux_records_test.go @@ -9,16 +9,27 @@ import ( ) // pastGrace is comfortably older than gcRecordGracePeriod; inGrace is newer. +// They are first-observed-orphaned times now, not record timestamps: grace runs +// from when the sweep first saw a slot orphaned, because a scan record's own +// scannedAt is restamped by every rescan and never ages. See +// TestAuxRecordOrphaned_RescannedRecordIsEventuallyCollected. var ( pastGrace = time.Now().Add(-gcRecordGracePeriod - 24*time.Hour) inGrace = time.Now().Add(-1 * time.Minute) ) +// orphanedSince builds the carried-forward orphan clock for a single slot, as +// if an earlier pass had first observed it orphaned at t. +func orphanedSince(recordKey string, t time.Time) map[string]time.Time { + return map[string]time.Time{recordKey: t} +} + 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" + recordKey = "io.atcr.hold.scan/def456" ) knownManifests := map[string]*manifestInfo{ @@ -35,61 +46,67 @@ func TestAuxRecordOrphaned(t *testing.T) { tests := []struct { name string manifestURI string - createdAt time.Time - wantOrphan bool - wantDID string + // orphanedAt is when an earlier pass first observed this slot orphaned. + // The zero value means no earlier pass did — this is a first sighting. + orphanedAt time.Time + wantOrphan bool + wantDID string }{ { name: "manifest deleted on reachable PDS is orphaned", manifestURI: deletedURI, - createdAt: pastGrace, + orphanedAt: pastGrace, wantOrphan: true, wantDID: "did:plc:alice", }, { name: "manifest still present is kept", manifestURI: knownURI, - createdAt: pastGrace, + orphanedAt: pastGrace, wantOrphan: false, }, { - name: "record inside grace window is kept", + name: "slot first seen orphaned inside the grace window is kept", + manifestURI: deletedURI, + orphanedAt: inGrace, + wantOrphan: false, + }, + { + name: "slot never observed orphaned before 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{}, + orphanedAt: pastGrace, wantOrphan: false, }, { name: "unparseable manifest URI is kept", manifestURI: "not-an-at-uri", - createdAt: pastGrace, + orphanedAt: pastGrace, wantOrphan: false, }, { name: "empty manifest URI is kept", manifestURI: "", - createdAt: pastGrace, + orphanedAt: pastGrace, wantOrphan: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - parts, orphan := auxRecordOrphaned(tt.manifestURI, tt.createdAt, auxOrphanState{ + since := map[string]time.Time{} + if !tt.orphanedAt.IsZero() { + since = orphanedSince(recordKey, tt.orphanedAt) + } + parts, orphan := auxRecordOrphaned(recordKey, tt.manifestURI, auxOrphanState{ knownManifests: knownManifests, knownDigests: knownDigests, fetchedUsers: fetchedUsers, + orphanSince: since, }) if orphan != tt.wantOrphan { t.Errorf("auxRecordOrphaned(%q) = %v, want %v", tt.manifestURI, orphan, tt.wantOrphan) @@ -107,22 +124,32 @@ func TestAuxRecordOrphaned(t *testing.T) { } } -// A record exactly at the grace boundary must be collected, not kept — the -// boundary test is `< gcRecordGracePeriod`, so equal-or-older is eligible. +// A slot first observed orphaned exactly a grace period ago 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" + const ( + uri = "at://did:plc:alice/io.atcr.manifest/gone" + recordKey = "io.atcr.hold.scan/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. + // between constructing the timestamp and evaluating the comparison. justPast := time.Now().Add(-gcRecordGracePeriod - time.Second) - if _, orphan := auxRecordOrphaned(uri, justPast, auxOrphanState{fetchedUsers: fetchedUsers}); !orphan { - t.Error("record just past the grace period should be orphaned") + if _, orphan := auxRecordOrphaned(recordKey, uri, auxOrphanState{ + fetchedUsers: fetchedUsers, + orphanSince: orphanedSince(recordKey, justPast), + }); !orphan { + t.Error("slot orphaned since just past the grace period should be collected") } justInside := time.Now().Add(-gcRecordGracePeriod + time.Minute) - if _, orphan := auxRecordOrphaned(uri, justInside, auxOrphanState{fetchedUsers: fetchedUsers}); orphan { - t.Error("record just inside the grace period should be kept") + if _, orphan := auxRecordOrphaned(recordKey, uri, auxOrphanState{ + fetchedUsers: fetchedUsers, + orphanSince: orphanedSince(recordKey, justInside), + }); orphan { + t.Error("slot orphaned since just inside the grace period should be kept") } } @@ -273,10 +300,16 @@ func TestDecodeAuxRecordBytes(t *testing.T) { } } -// 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" +// A record whose timestamp doesn't parse must decode without error and yield +// the zero time. The timestamp is diagnostic only — it is logged beside an +// orphan, not used to judge one — so a malformed date must neither fail the +// decode (which would skip the record) nor change the verdict: the slot is +// judged purely on how long it has looked orphaned. +func TestDecodeAuxRecordBytesBadTimestamp(t *testing.T) { + const ( + manifestURI = "at://did:plc:alice/io.atcr.manifest/abc123" + recordKey = "io.atcr.hold.scan/abc123" + ) scan := &atproto.ScanRecord{ Type: atproto.ScanCollection, @@ -300,7 +333,21 @@ func TestDecodeAuxRecordBytesBadTimestampIsKept(t *testing.T) { } fetchedUsers := map[string]bool{"did:plc:alice": true} - if _, orphan := auxRecordOrphaned(uri, ts, auxOrphanState{fetchedUsers: fetchedUsers}); orphan { - t.Error("record with an unparseable timestamp should be kept, not orphaned") + + // Not yet observed orphaned for long enough: kept, as any record would be. + if _, orphan := auxRecordOrphaned(recordKey, uri, auxOrphanState{ + fetchedUsers: fetchedUsers, + orphanSince: orphanedSince(recordKey, inGrace), + }); orphan { + t.Error("record inside the grace window should be kept regardless of its timestamp") + } + + // Observed orphaned across a full grace period: collected. A record does + // not become immortal by carrying a date nobody can parse. + if _, orphan := auxRecordOrphaned(recordKey, uri, auxOrphanState{ + fetchedUsers: fetchedUsers, + orphanSince: orphanedSince(recordKey, pastGrace), + }); !orphan { + t.Error("an unparseable timestamp must not exempt a sustained orphan from collection") } } diff --git a/pkg/hold/gc/config.go b/pkg/hold/gc/config.go index c848e41..26e18ee 100644 --- a/pkg/hold/gc/config.go +++ b/pkg/hold/gc/config.go @@ -21,14 +21,24 @@ const ( // 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. + // gcRecordGracePeriod is how long a record must have looked collectable + // 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. + // written before the manifest reaches the user's PDS, so a record that has + // looked orphaned for less 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. + // + // The two sweeps start the clock differently, because only one of them has + // a write time it can trust. Layer records have unique per-write TID rkeys, + // so the rkey dates the write and the sweep measures from it. Scan and + // image-config records live at a rkey derived from the manifest digest and + // are rewritten in place — a rescan restamps io.atcr.hold.scan's only + // timestamp — so their sweep measures from the first pass that observed + // them orphaned instead. See auxOrphanState.graceElapsed. gcRecordGracePeriod = 24 * time.Hour // maxPreviewAgeForDelete bounds how stale a preview may be when the admin diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index 79e18fa..54890db 100644 --- a/pkg/hold/gc/gc.go +++ b/pkg/hold/gc/gc.go @@ -65,6 +65,99 @@ type auxOrphanState struct { digestOwners map[string]map[string]bool // fetchedUsers are the DIDs whose PDS answered completely this run. fetchedUsers map[string]bool + + // now is the wall clock for this analysis pass, held once so every record + // in the pass is judged against the same instant. The zero value means + // time.Now(), which is what unit tests that don't care about the clock get. + now time.Time + // orphanSince carries forward, per auxRecordKey slot, the first time an + // earlier pass in this process observed that slot orphaned. Reading it is + // what makes grace measure orphanhood rather than record age. + orphanSince map[string]time.Time + // orphanObserved collects this pass's observations, keyed the same way, for + // the caller to carry into the next pass. A nil map means "carry nothing + // forward", which fails safe: no slot accrues age, so nothing is collected. + orphanObserved map[string]time.Time +} + +// auxRecordKey identifies one auxiliary record slot for the orphan clock. +// +// Deliberately collection+rkey with no CID: the clock times how long the SLOT +// has been orphaned, and a rewrite that leaves it orphaned — a rescan +// restamping io.atcr.hold.scan — must not restart it. A rewrite that makes the +// slot live again needs no special handling here; the orphan test stops +// observing it and commit drops the entry. +func auxRecordKey(collection, rkey string) string { + return collection + "/" + rkey +} + +// graceElapsed records that recordKey looks orphaned as of this pass, and +// reports whether it has looked that way for at least gcRecordGracePeriod. +// +// Every uncertain input starts the clock at now rather than expiring it: an +// unrecorded slot (first pass, or first pass after a restart), a zero time, and +// a time in the future all mean "we cannot show this has been orphaned for a +// full grace period", so the record survives at least one more pass. +// +// Call this LAST, after a record has already failed every other test. Observing +// a slot that is not actually orphaned accrues age against it, and the entry +// would then already be stale the moment its manifest genuinely disappeared. +func (s auxOrphanState) graceElapsed(recordKey string) bool { + now := s.now + if now.IsZero() { + now = time.Now() + } + + since, ok := s.orphanSince[recordKey] + if !ok || since.IsZero() || since.After(now) { + since = now + } + if s.orphanObserved != nil { + s.orphanObserved[recordKey] = since + } + return now.Sub(since) >= gcRecordGracePeriod +} + +// auxOrphanClock is the process-lifetime memory behind that calculation: which +// auxiliary record slots looked orphaned, and since when. +// +// It is in-memory only, so a restart forgets every observation and each +// surviving orphan starts its grace period over. That is the conservative +// direction — a restart delays collection and never advances it — and it is +// why commit replaces rather than merges. +type auxOrphanClock struct { + mu sync.Mutex + since map[string]time.Time +} + +// snapshot copies the recorded first-observation times for one analysis pass. +func (c *auxOrphanClock) snapshot() map[string]time.Time { + c.mu.Lock() + defer c.mu.Unlock() + + out := make(map[string]time.Time, len(c.since)) + for k, v := range c.since { + out[k] = v + } + return out +} + +// commit replaces the recorded times with a completed pass's observations. +// +// Replacing rather than merging is the point: a slot the pass did not observe +// orphaned loses its accrued age. That covers the record coming back to life, +// the record being deleted, and the record's owner being unreachable — the last +// of which is why this is a replace. An owner we could not reach is an owner we +// cannot judge, and the sweep's standing rule is that an unreachable PDS never +// contributes to a deletion, not even by letting a clock run in the background. +// +// Only call this for a pass that completed. A pass that failed partway has +// observed only a prefix of the collection, and committing it would reset the +// clock on everything past that point. +func (c *auxOrphanClock) commit(observed map[string]time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + c.since = observed } // recordDigestOwner notes that the DID in manifestURI owns a manifest at that @@ -260,6 +353,11 @@ type GarbageCollector struct { // still serving on the predecessor's behalf. predecessorCache map[string]bool + // auxOrphans times how long each scan / image-config record slot has looked + // orphaned, across analysis passes. See auxOrphanClock: it is the grace + // clock for those two collections, and it lives only as long as the process. + auxOrphans auxOrphanClock + // predecessorUnresolved holds the DIDs whose predecessor status could not be // determined during the current analysis. It exists only so that one // unreachable hold costs a single 5s timeout per run rather than one per @@ -1022,22 +1120,40 @@ 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. + // digest rather than TID, 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. + // + // Their blob references are NOT part of the referenced set this sweep + // builds. io.atcr.hold.scan carries sbomBlob and vulnReportBlob, but those + // live in the hold's own PDS blob store at /repos//blobs/, + // while the blob sweep walks the registry prefix /docker/registry/v2/blobs. + // The two spaces do not overlap, so deleting a scan record here neither + // frees its SBOM nor risks orphaning a blob this sweep could delete. That + // blob space is currently never collected at all; a sweep for it would have + // to read these two fields before the record goes. gc.setProgress("records", "Scanning scan and image config records...", gc.operationType) auxState := auxOrphanState{ knownManifests: knownManifests, knownDigests: knownDigests, digestOwners: digestOwners, fetchedUsers: fetchedUsers, + now: time.Now(), + orphanSince: gc.auxOrphans.snapshot(), + orphanObserved: make(map[string]time.Time), } for _, collection := range []string{atproto.ScanCollection, atproto.ImageConfigCollection} { if err := gc.scanAuxRecords(ctx, collection, auxState, result); err != nil { + // Deliberately not committing: a pass that stopped partway saw only + // part of the collection, and committing it would reset the grace + // clock on every slot it never reached. return nil, fmt.Errorf("scan %s records: %w", collection, err) } } + // Both collections walked end to end, so this pass's observations are a + // complete picture and can replace the previous one. + gc.auxOrphans.commit(auxState.orphanObserved) // Step 4: Identify missing layer records (uncovered manifest+layer pairs) for _, m := range knownManifests { @@ -1067,10 +1183,13 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult // 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. +// Grace comes from state's orphan clock, not from the record. These +// collections use deterministic digest rkeys rather than TIDs, so there is no +// write time in the rkey to read, and the timestamps in the body cannot stand +// in for one: io.atcr.hold.scan's scannedAt is restamped by every rescan, so on +// a hold whose rescan_interval is shorter than gcRecordGracePeriod it never +// ages past grace and the record is immortal. The clock times orphanhood +// instead, which is what grace was always meant to measure. func (gc *GarbageCollector) scanAuxRecords( ctx context.Context, collection string, @@ -1094,14 +1213,14 @@ func (gc *GarbageCollector) scanAuxRecords( scanned++ result.totalRecords++ - manifestURI, createdAt, recCID, err := gc.decodeAuxRecord(ctx, collection, rec) + manifestURI, writtenAt, 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, state) + parts, isOrphan := auxRecordOrphaned(auxRecordKey(collection, rec.Rkey), manifestURI, state) if !isOrphan { continue } @@ -1122,8 +1241,13 @@ func (gc *GarbageCollector) scanAuxRecords( UserDID: parts.DID, }) } + // writtenAt is the record body's own timestamp (scannedAt / + // createdAt). It is diagnostic only — a scan record's is restamped + // by every rescan, which is exactly why it no longer gates grace. gc.logger.Debug("Found orphaned record", - "collection", collection, "rkey", rec.Rkey, "manifest", manifestURI) + "collection", collection, "rkey", rec.Rkey, "manifest", manifestURI, + "recordWrittenAt", writtenAt, + "orphanedSince", state.orphanSince[auxRecordKey(collection, rec.Rkey)]) } if nextCursor == "" { @@ -1138,17 +1262,20 @@ func (gc *GarbageCollector) scanAuxRecords( } // 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. +// collected, given the slot it occupies and the manifest it names. 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, state auxOrphanState) (*atURIParts, bool) { - if createdAt.IsZero() || time.Since(createdAt) < gcRecordGracePeriod { - return nil, false - } +// Every uncertain case resolves to "keep": a record whose manifest URI is +// unparseable, one whose owning PDS was unreachable this run, one a co-owner +// might still need, and one that has not yet looked orphaned for a full grace +// period. Only a record whose every reachable owner demonstrably lacks the +// manifest, sustained across grace, is orphaned. +// +// The grace test comes last, and deliberately so. It both reads and advances +// the orphan clock, so it must only ever see records that have already failed +// every other test — a record judged live must not accrue orphan age. +func auxRecordOrphaned(recordKey, manifestURI string, state auxOrphanState) (*atURIParts, bool) { if _, known := state.knownManifests[manifestURI]; known { return nil, false } @@ -1181,6 +1308,14 @@ func auxRecordOrphaned(manifestURI string, createdAt time.Time, state auxOrphanS if state.knownDigests[digest] { return nil, false } + + // The record looks orphaned right now. Grace runs from the first pass that + // saw it this way — not from any timestamp inside the record, which a + // rescan would reset — so a slot that has only just started looking + // orphaned survives to be re-judged next pass. + if !state.graceElapsed(recordKey) { + return nil, false + } return parts, true } diff --git a/pkg/hold/gc/orphan_regression_test.go b/pkg/hold/gc/orphan_regression_test.go index 3840fd7..b82a367 100644 --- a/pkg/hold/gc/orphan_regression_test.go +++ b/pkg/hold/gc/orphan_regression_test.go @@ -21,6 +21,12 @@ import ( const regressionDigest = "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f" +// regressionScanKey is the orphan-clock slot for that digest's scan record. +// These tests are about WHICH records the sweep judges collectable, not about +// how long it waits, so they seed the clock past grace (see orphanedSince) and +// let the co-ownership checks do the work. +var regressionScanKey = auxRecordKey(atproto.ScanCollection, atproto.ScanRecordKey(regressionDigest)) + // 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 @@ -62,12 +68,15 @@ func TestAuxRecordOrphaned_DigestSharedAcrossUsers(t *testing.T) { 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{ + // The record at hex(digest) names Bob, the last writer, and has looked that + // way since well before the grace period, so only the co-ownership check + // can save it. + _, isOrphan := auxRecordOrphaned(regressionScanKey, bobURI, auxOrphanState{ knownManifests: knownManifests, knownDigests: knownDigests, digestOwners: digestOwners, fetchedUsers: fetchedUsers, + orphanSince: orphanedSince(regressionScanKey, pastGrace), }) if isOrphan { @@ -105,11 +114,12 @@ func TestAuxRecordOrphaned_DigestUniqueToUser(t *testing.T) { extractDigestFromManifestURI(bobURI): {"did:plc:bob": true}, } - parts, isOrphan := auxRecordOrphaned(bobURI, pastGrace, auxOrphanState{ + parts, isOrphan := auxRecordOrphaned(regressionScanKey, bobURI, auxOrphanState{ knownManifests: knownManifests, knownDigests: knownDigests, digestOwners: digestOwners, fetchedUsers: fetchedUsers, + orphanSince: orphanedSince(regressionScanKey, pastGrace), }) if !isOrphan { t.Fatal("record whose only referencing manifest was deleted should be collected") @@ -292,9 +302,10 @@ func TestAuxRecordOrphaned_UnreachableCoOwnerIsKept(t *testing.T) { digest: {"did:plc:alice": true, "did:plc:bob": true}, }, fetchedUsers: map[string]bool{"did:plc:bob": true}, + orphanSince: orphanedSince(regressionScanKey, pastGrace), } - if _, orphan := auxRecordOrphaned(bobURI, pastGrace, state); orphan { + if _, orphan := auxRecordOrphaned(regressionScanKey, bobURI, 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) @@ -316,9 +327,10 @@ func TestAuxRecordOrphaned_AllOwnersReachableAndGoneIsCollected(t *testing.T) { digest: {"did:plc:alice": true, "did:plc:bob": true}, }, fetchedUsers: map[string]bool{"did:plc:alice": true, "did:plc:bob": true}, + orphanSince: orphanedSince(regressionScanKey, pastGrace), } - parts, orphan := auxRecordOrphaned(bobURI, pastGrace, state) + parts, orphan := auxRecordOrphaned(regressionScanKey, bobURI, state) if !orphan { t.Fatal("record whose every owner was reached and no longer holds the manifest should be collected") }