From 1a2ea93c8d761ae84a8ec72a7610a86a29b52358 Mon Sep 17 00:00:00 2001 From: ztzhou Date: Mon, 9 Nov 2026 20:04:13 +0800 Subject: [PATCH 1/2] Prevent retention from clearing recently served artifacts --- internal/database/hits.go | 4 ++++ internal/database/retention.go | 32 +++++++++++++++++++++++++++++ internal/database/retention_test.go | 23 +++++++++++++++++++++ 3 files changed, 59 insertions(+) diff --git a/internal/database/hits.go b/internal/database/hits.go index 3609b5f9..43f29235 100644 --- a/internal/database/hits.go +++ b/internal/database/hits.go @@ -26,6 +26,7 @@ type hitEntry struct { type metadataHitKey struct{ ecosystem, name string } type hitBatch struct { + flushMu sync.Mutex mu sync.Mutex pending map[hitKey]hitEntry metadata map[metadataHitKey]time.Time @@ -105,6 +106,9 @@ func (db *DB) BatchHits(interval time.Duration, logger *slog.Logger) { // fails they are put back for the next flush, so a failed write delays // counts rather than losing them. func (db *DB) flushHits() error { + db.hits.flushMu.Lock() + defer db.hits.flushMu.Unlock() + metadataErr := db.flushMetadataHits() pending := db.hits.take() if len(pending) == 0 { diff --git a/internal/database/retention.go b/internal/database/retention.go index 0036362c..efab4c50 100644 --- a/internal/database/retention.go +++ b/internal/database/retention.go @@ -83,6 +83,16 @@ func (db *DB) GetRetentionCandidates(fromID, toID int64, cutoff time.Time) ([]Re // path and a later sweep tries again, instead of the object staying in // storage with nothing left pointing at it. func (db *DB) ClearExpiredArtifact(versionPURL, filename, storagePath string, cutoff time.Time) (bool, error) { + if db.hits != nil { + db.hits.flushMu.Lock() + defer db.hits.flushMu.Unlock() + db.hits.mu.Lock() + defer db.hits.mu.Unlock() + if err := db.flushHitsLocked(); err != nil { + return false, err + } + } + tx, err := db.Beginx() if err != nil { return false, err @@ -108,6 +118,28 @@ func (db *DB) ClearExpiredArtifact(versionPURL, filename, storagePath string, cu return true, tx.Commit() } +// flushHitsLocked writes the current artifact-hit batch while the caller +// holds both hit locks, serializing it with background flushes and new hits. +func (db *DB) flushHitsLocked() error { + pending := db.hits.pending + db.hits.pending = map[hitKey]hitEntry{} + if len(pending) == 0 { + return nil + } + if err := db.writeHits(pending); err != nil { + for k, e := range pending { + cur := db.hits.pending[k] + cur.count += e.count + if e.last.After(cur.last) { + cur.last = e.last + } + db.hits.pending[k] = cur + } + return err + } + return nil +} + // MarkArtifactMirrored records that a mirror run found the artifact cached, // which keeps it from retention without counting as a cache hit. func (db *DB) MarkArtifactMirrored(versionPURL, filename string) error { diff --git a/internal/database/retention_test.go b/internal/database/retention_test.go index 23820110..89bc5643 100644 --- a/internal/database/retention_test.go +++ b/internal/database/retention_test.go @@ -221,6 +221,29 @@ func TestClearExpiredArtifact(t *testing.T) { }) } +func TestClearExpiredArtifactFlushesHitsBufferedAfterSweepStart(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + old := now.Add(-48 * time.Hour) + cutoff := now.Add(-24 * time.Hour) + seedRetentionArtifact(t, db, "npm", "served", old, old) + db.BatchHits(time.Hour, nil) + + if err := db.FlushHits(); err != nil { + t.Fatalf("initial FlushHits: %v", err) + } + if err := db.RecordArtifactHit("pkg:npm/served@1.0.0", "served.tgz"); err != nil { + t.Fatalf("RecordArtifactHit: %v", err) + } + + cleared, err := db.ClearExpiredArtifact("pkg:npm/served@1.0.0", "served.tgz", + "npm/served/1.0.0/served.tgz", cutoff) + if err != nil || cleared { + t.Fatalf("ClearExpiredArtifact(served) = %v, %v, want false, nil", cleared, err) + } + }) +} + // When the object can't be queued for deletion, the clear is rolled back: // the record keeps pointing at the object, so a later sweep retries it // rather than leaving a file in storage that nothing refers to. From 5500a6907563945f7a3ffff198941dd39a9f5841 Mon Sep 17 00:00:00 2001 From: ztzhou Date: Fri, 13 Nov 2026 15:09:51 +0800 Subject: [PATCH 2/2] Check pending hits before clearing artifacts --- internal/database/hits.go | 4 --- internal/database/retention.go | 41 +++++++---------------------- internal/database/retention_test.go | 2 +- 3 files changed, 10 insertions(+), 37 deletions(-) diff --git a/internal/database/hits.go b/internal/database/hits.go index 43f29235..3609b5f9 100644 --- a/internal/database/hits.go +++ b/internal/database/hits.go @@ -26,7 +26,6 @@ type hitEntry struct { type metadataHitKey struct{ ecosystem, name string } type hitBatch struct { - flushMu sync.Mutex mu sync.Mutex pending map[hitKey]hitEntry metadata map[metadataHitKey]time.Time @@ -106,9 +105,6 @@ func (db *DB) BatchHits(interval time.Duration, logger *slog.Logger) { // fails they are put back for the next flush, so a failed write delays // counts rather than losing them. func (db *DB) flushHits() error { - db.hits.flushMu.Lock() - defer db.hits.flushMu.Unlock() - metadataErr := db.flushMetadataHits() pending := db.hits.take() if len(pending) == 0 { diff --git a/internal/database/retention.go b/internal/database/retention.go index efab4c50..1c369335 100644 --- a/internal/database/retention.go +++ b/internal/database/retention.go @@ -77,19 +77,18 @@ func (db *DB) GetRetentionCandidates(fromID, toID int64, cutoff time.Time) ([]Re } // ClearExpiredArtifact marks an artifact uncached like ClearArtifactCache, -// but only while it is still expired before cutoff, so an artifact served -// between the scan and the clear stays cached. It queues storagePath for -// deletion in the same transaction: if queueing fails, the record keeps its -// path and a later sweep tries again, instead of the object staying in -// storage with nothing left pointing at it. +// but only while it is still expired before cutoff. It checks pending hits as +// well as persisted access times, so a hit buffered after the scan keeps the +// artifact cached. It queues storagePath for deletion in the same transaction: +// if queueing fails, the record keeps its path and a later sweep retries, +// instead of leaving an object in storage with no record pointing to it. func (db *DB) ClearExpiredArtifact(versionPURL, filename, storagePath string, cutoff time.Time) (bool, error) { if db.hits != nil { - db.hits.flushMu.Lock() - defer db.hits.flushMu.Unlock() db.hits.mu.Lock() - defer db.hits.mu.Unlock() - if err := db.flushHitsLocked(); err != nil { - return false, err + pending, ok := db.hits.pending[hitKey{versionPURL, filename}] + db.hits.mu.Unlock() + if ok && !pending.last.Before(cutoff) { + return false, nil } } @@ -118,28 +117,6 @@ func (db *DB) ClearExpiredArtifact(versionPURL, filename, storagePath string, cu return true, tx.Commit() } -// flushHitsLocked writes the current artifact-hit batch while the caller -// holds both hit locks, serializing it with background flushes and new hits. -func (db *DB) flushHitsLocked() error { - pending := db.hits.pending - db.hits.pending = map[hitKey]hitEntry{} - if len(pending) == 0 { - return nil - } - if err := db.writeHits(pending); err != nil { - for k, e := range pending { - cur := db.hits.pending[k] - cur.count += e.count - if e.last.After(cur.last) { - cur.last = e.last - } - db.hits.pending[k] = cur - } - return err - } - return nil -} - // MarkArtifactMirrored records that a mirror run found the artifact cached, // which keeps it from retention without counting as a cache hit. func (db *DB) MarkArtifactMirrored(versionPURL, filename string) error { diff --git a/internal/database/retention_test.go b/internal/database/retention_test.go index 89bc5643..6a76ec59 100644 --- a/internal/database/retention_test.go +++ b/internal/database/retention_test.go @@ -221,7 +221,7 @@ func TestClearExpiredArtifact(t *testing.T) { }) } -func TestClearExpiredArtifactFlushesHitsBufferedAfterSweepStart(t *testing.T) { +func TestClearExpiredArtifactSkipsHitBufferedAfterSweepStart(t *testing.T) { runWithBothDatabases(t, func(t *testing.T, db *DB) { now := time.Now() old := now.Add(-48 * time.Hour)