From c0320c7c87a5be942bd8be893b73c3b07a2ea75f Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sat, 17 Oct 2026 23:20:45 +0200 Subject: [PATCH 1/3] Clearing a record and queueing its file for deletion now happen in one transaction, so a failed queue write keeps the record instead of leaving a file nothing points at. Eviction drops the queue entry again once it has deleted the file itself. --- internal/database/pending_deletes_test.go | 44 +++++++++++++- internal/database/queries.go | 73 ++++++++++++++++------- internal/server/eviction.go | 8 +-- internal/server/eviction_test.go | 55 +++++++++++++++++ 4 files changed, 151 insertions(+), 29 deletions(-) diff --git a/internal/database/pending_deletes_test.go b/internal/database/pending_deletes_test.go index 9af6c48d..f1245571 100644 --- a/internal/database/pending_deletes_test.go +++ b/internal/database/pending_deletes_test.go @@ -79,7 +79,7 @@ func TestUpsertArtifactSkipsRecordThatMoved(t *testing.T) { UpstreamURL: "https://example.com/" + pendingFilename, StoragePath: sql.NullString{String: "npm/pending/1.0.0/c/pending-1.0.0.tgz", Valid: true}, } - applied, err := db.upsertArtifactFrom(late, read) + applied, err := db.upsertArtifactFrom(db, late, read) if err != nil { t.Fatalf("upsertArtifactFrom: %v", err) } @@ -91,7 +91,7 @@ func TestUpsertArtifactSkipsRecordThatMoved(t *testing.T) { } missing := &Artifact{VersionPURL: "pkg:npm/pending@2.0.0", Filename: pendingFilename, UpstreamURL: "u"} - applied, err = db.upsertArtifactFrom(missing, sql.NullString{}) + applied, err = db.upsertArtifactFrom(db, missing, sql.NullString{}) if err != nil || !applied { t.Errorf("insert of a new record: applied=%v err=%v", applied, err) } @@ -173,6 +173,46 @@ func TestDiscardArtifactClearsAndQueues(t *testing.T) { }) } +// When the replaced path can't be queued, the write is rolled back, so the +// record keeps pointing at the old object instead of orphaning it. +func TestUpsertArtifactRollsBackWhenQueueingFails(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + upsertPendingArtifact(t, db, "npm/pending/1.0.0/a/pending-1.0.0.tgz") + if _, err := db.Exec(`DROP TABLE pending_deletes`); err != nil { + t.Fatalf("dropping pending_deletes: %v", err) + } + + err := db.UpsertArtifact(&Artifact{ + VersionPURL: pendingVersionPURL, + Filename: pendingFilename, + UpstreamURL: "https://example.com/" + pendingFilename, + StoragePath: sql.NullString{String: "npm/pending/1.0.0/b/pending-1.0.0.tgz", Valid: true}, + }) + if err == nil { + t.Fatal("UpsertArtifact succeeded without queueing the replaced path") + } + if got := recordedPath(t, db).String; got != "npm/pending/1.0.0/a/pending-1.0.0.tgz" { + t.Errorf("record points at %q, want the old path kept", got) + } + }) +} + +func TestDiscardArtifactRollsBackWhenQueueingFails(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + upsertPendingArtifact(t, db, "npm/pending/1.0.0/a/pending-1.0.0.tgz") + if _, err := db.Exec(`DROP TABLE pending_deletes`); err != nil { + t.Fatalf("dropping pending_deletes: %v", err) + } + + if err := db.DiscardArtifact(pendingVersionPURL, pendingFilename, "npm/pending/1.0.0/a/pending-1.0.0.tgz"); err == nil { + t.Fatal("DiscardArtifact succeeded without queueing the path") + } + if got := recordedPath(t, db).String; got != "npm/pending/1.0.0/a/pending-1.0.0.tgz" { + t.Errorf("record points at %q, want it kept", got) + } + }) +} + func TestGetDuePendingDeletes(t *testing.T) { runWithBothDatabases(t, func(t *testing.T, db *DB) { for _, path := range []string{"old", "referenced"} { diff --git a/internal/database/queries.go b/internal/database/queries.go index cd7c51ce..ea63f15a 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -9,6 +9,7 @@ import ( "github.com/git-pkgs/artifacts" "github.com/git-pkgs/purl" + "github.com/jmoiron/sqlx" "github.com/opencontainers/go-digest" ) @@ -340,26 +341,42 @@ func (db *DB) UpsertArtifact(a *Artifact) error { if err := db.Get(&previous, query, a.VersionPURL, a.Filename); err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("reading artifact: %w", err) } - applied, err := db.upsertArtifactFrom(a, previous) + applied, err := db.upsertArtifactQueuing(a, previous) if err != nil { return err } - if !applied { - continue - } - if previous.Valid && previous.String != a.StoragePath.String { - if err := db.QueuePendingDelete(previous.String); err != nil { - return fmt.Errorf("queueing replaced artifact: %w", err) - } + if applied { + return nil } - return nil } return errors.New("upserting artifact: record kept changing") } +// upsertArtifactQueuing writes a as upsertArtifactFrom does and queues the +// path it replaced in the same transaction, so a failed queue leaves the +// record on the old path instead of orphaning it. +func (db *DB) upsertArtifactQueuing(a *Artifact, previous sql.NullString) (bool, error) { + tx, err := db.Beginx() + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + + applied, err := db.upsertArtifactFrom(tx, a, previous) + if err != nil || !applied { + return false, err + } + if previous.Valid && previous.String != a.StoragePath.String { + if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), previous.String, time.Now().UTC()); err != nil { + return false, fmt.Errorf("queueing replaced artifact: %w", err) + } + } + return true, tx.Commit() +} + // upsertArtifactFrom writes a if the record is absent or still holds // previous, and reports whether it did. -func (db *DB) upsertArtifactFrom(a *Artifact, previous sql.NullString) (bool, error) { +func (db *DB) upsertArtifactFrom(ex sqlx.Execer, a *Artifact, previous sql.NullString) (bool, error) { now := time.Now() var query string @@ -401,7 +418,7 @@ func (db *DB) upsertArtifactFrom(a *Artifact, previous sql.NullString) (bool, er ` } - res, err := db.Exec(query, + res, err := ex.Exec(query, a.VersionPURL, a.Filename, a.UpstreamURL, a.StoragePath, a.ContentHash, a.Size, a.ContentType, inUTC(a.FetchedAt), a.HitCount, inUTC(a.LastAccessedAt), now, now, a.sourceOrUpstream(), a.Manifest, previous, @@ -482,33 +499,43 @@ func (db *DB) GetCachedArtifactCount() (int64, error) { } // ClearArtifactCache marks an artifact uncached if its record still points at -// storagePath, and reports whether it did. A record a newer fetch has moved -// elsewhere is left alone, so a clear never orphans the object that fetch -// committed. +// storagePath, queues storagePath for deletion in the same transaction, and +// reports whether it did. A record a newer fetch has moved elsewhere is left +// alone, so a clear never orphans the object that fetch committed. If +// queueing fails, the record keeps its path. func (db *DB) ClearArtifactCache(versionPURL, filename, storagePath string) (bool, error) { + tx, err := db.Beginx() + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + query := db.Rebind(` UPDATE artifacts SET storage_path = NULL, content_hash = NULL, size = NULL, content_type = NULL, fetched_at = NULL, updated_at = ? WHERE version_purl = ? AND filename = ? AND storage_path = ? `) - res, err := db.Exec(query, time.Now(), versionPURL, filename, storagePath) + res, err := tx.Exec(query, time.Now(), versionPURL, filename, storagePath) if err != nil { return false, err } n, err := res.RowsAffected() - return n > 0, err + if err != nil || n == 0 { + return false, err + } + if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), storagePath, time.Now().UTC()); err != nil { + return false, fmt.Errorf("queueing %s for deletion: %w", storagePath, err) + } + return true, tx.Commit() } -// DiscardArtifact clears the record as ClearArtifactCache does and queues -// storagePath for deletion, for callers that must not delete an object another +// DiscardArtifact clears the record and queues storagePath as +// ClearArtifactCache does, for callers that must not delete an object another // request may be reading. func (db *DB) DiscardArtifact(versionPURL, filename, storagePath string) error { - cleared, err := db.ClearArtifactCache(versionPURL, filename, storagePath) - if err != nil || !cleared { - return err - } - return db.QueuePendingDelete(storagePath) + _, err := db.ClearArtifactCache(versionPURL, filename, storagePath) + return err } // QueuePendingDelete queues a storage path no record points at any more. diff --git a/internal/server/eviction.go b/internal/server/eviction.go index 7fe108a4..fe69db23 100644 --- a/internal/server/eviction.go +++ b/internal/server/eviction.go @@ -120,6 +120,7 @@ func evictBatch(ctx context.Context, db *database.DB, store storage.Storage, log // Clear before deleting: a record a newer fetch moved has left this // path queued, and a request may still open it within the grace period. + // The clear queues the path too, so it is never left unrecorded. recordCleared, err := db.ClearArtifactCache(art.VersionPURL, art.Filename, path) if err != nil { logger.Warn("eviction: failed to clear artifact record", @@ -131,10 +132,9 @@ func evictBatch(ctx context.Context, db *database.DB, store storage.Storage, log } if err := store.Delete(ctx, path); err != nil { - logger.Warn("eviction: failed to delete from storage, queueing it", "path", path, "error", err) - if err := db.QueuePendingDelete(path); err != nil { - logger.Warn("eviction: failed to queue object for deletion", "path", path, "error", err) - } + logger.Warn("eviction: failed to delete from storage, leaving it queued", "path", path, "error", err) + } else if err := db.RemovePendingDelete(path); err != nil { + logger.Warn("eviction: failed to dequeue deleted object", "path", path, "error", err) } if art.Size.Valid { diff --git a/internal/server/eviction_test.go b/internal/server/eviction_test.go index 15355864..87302961 100644 --- a/internal/server/eviction_test.go +++ b/internal/server/eviction_test.go @@ -487,3 +487,58 @@ func TestEvictBatch_SkipsRecordThatMoved(t *testing.T) { t.Errorf("queue = %v, want the old path kept for reclaim", got) } } + +// TestEvictBatch_KeepsRecordWhenQueueingFails evicts while the queue refuses +// writes. The record must keep its path and the object stay, or a crash before +// the delete would leave a file nothing points at. +func TestEvictBatch_KeepsRecordWhenQueueingFails(t *testing.T) { + db, store := setupEvictionTest(t) + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + seedArtifact(t, ctx, db, store, "unqueued", 1000, time.Now().Add(-time.Hour)) + + rows, err := db.GetLeastRecentlyUsedArtifacts(evictionBatch) + if err != nil || len(rows) != 1 { + t.Fatalf("reading LRU rows: %v, %v", rows, err) + } + if _, err := db.Exec(`CREATE TRIGGER refuse_queue BEFORE INSERT ON pending_deletes + BEGIN SELECT RAISE(ABORT, 'queue refused'); END`); err != nil { + t.Fatalf("creating trigger: %v", err) + } + + cleared, freed := evictBatch(ctx, db, store, logger, rows, 0, 1000) + if cleared != 0 || freed != 0 { + t.Errorf("cleared %d records freeing %d bytes, want nothing", cleared, freed) + } + path := rows[0].StoragePath.String + record, err := db.GetArtifact(rows[0].VersionPURL, rows[0].Filename) + if err != nil || record == nil || record.StoragePath.String != path { + t.Fatalf("record = %+v (err %v), want it kept at %q", record, err, path) + } + if ok, _ := store.Exists(ctx, path); !ok { + t.Error("object deleted although its record was kept") + } +} + +// TestEvictBatch_DequeuesDeletedObjects checks that an object eviction deleted +// does not stay in the queue for reclaim to delete again. +func TestEvictBatch_DequeuesDeletedObjects(t *testing.T) { + db, store := setupEvictionTest(t) + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + seedArtifact(t, ctx, db, store, "gone", 1000, time.Now().Add(-time.Hour)) + + rows, err := db.GetLeastRecentlyUsedArtifacts(evictionBatch) + if err != nil || len(rows) != 1 { + t.Fatalf("reading LRU rows: %v, %v", rows, err) + } + if cleared, _ := evictBatch(ctx, db, store, logger, rows, 0, 1000); cleared != 1 { + t.Fatalf("cleared %d records, want 1", cleared) + } + if ok, _ := store.Exists(ctx, rows[0].StoragePath.String); ok { + t.Error("object still in storage") + } + if got := queuedPaths(t, db); len(got) != 0 { + t.Errorf("queue = %v, want it empty", got) + } +} From 24acee273118d841911d9b04998dd6d18cc88e75 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 26 Oct 2026 08:49:50 +0100 Subject: [PATCH 2/3] When a refetched file can be neither recorded nor queued for deletion, it is still served to everyone sharing the fetch and deleted once the last of them has closed it, so it no longer stays in storage with nothing pointing at it. A record that committed despite the reported error keeps its file. --- internal/handler/handler.go | 134 +++++++++++- internal/handler/unrecorded_test.go | 304 ++++++++++++++++++++++++++++ 2 files changed, 428 insertions(+), 10 deletions(-) create mode 100644 internal/handler/unrecorded_test.go diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 62d63aec..dd880dba 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -14,6 +14,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/git-pkgs/artifacts" @@ -617,14 +618,79 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil p.Logger.Warn("failed to update cache database", "error", err) // Continue anyway - we have the file. Queue it for deletion in case // no record points at it; the queue skips it while one does. - if qErr := p.DB.QueuePendingDelete(storagePath); qErr != nil { - p.Logger.Warn("failed to queue unrecorded artifact for deletion", "path", storagePath, "error", qErr) + if qErr := p.DB.QueuePendingDelete(storagePath); qErr != nil && !p.recordedAt(versionPURL, filename, storagePath) { + p.Logger.Warn("failed to queue unrecorded artifact for deletion, deleting it once served", + "path", storagePath, "error", qErr) + return sharedArtifact, storagePath, errUnrecorded } } return sharedArtifact, storagePath, nil } +// recordedAt reports whether the artifact's record points at storagePath. A +// commit can apply and still report an error, so storeArtifact checks before +// treating an object as unrecorded. A failed lookup counts as not recorded. +func (p *Proxy) recordedAt(versionPURL, filename, storagePath string) bool { + a, err := p.DB.GetArtifact(versionPURL, filename) + return err == nil && a != nil && a.StoragePath.Valid && a.StoragePath.String == storagePath +} + +// errUnrecorded comes back from storeArtifact together with a usable artifact +// and path when neither a record nor a queue entry could be written for it. +// Nothing else would ever delete the object, so the fetch serves it to +// everyone sharing it and deletes it when the last of them is done. +var errUnrecorded = errors.New("stored artifact has no record") + +// unrecordedObject deletes an unrecorded object once every caller sharing +// its fetch has released it. +type unrecordedObject struct { + refs atomic.Int32 + delete func() +} + +func (u *unrecordedObject) release() { + if u.refs.Add(-1) == 0 { + u.delete() + } +} + +// releaseOnClose releases an unrecorded object once its reader is closed. +type releaseOnClose struct { + io.ReadCloser + release func() +} + +func (r *releaseOnClose) Close() error { + err := r.ReadCloser.Close() + r.release() + return err +} + +// seekableReleaseOnClose keeps range requests working on a seekable reader. +type seekableReleaseOnClose struct { + releaseOnClose + io.Seeker +} + +// serveUnrecorded gives one caller its own reader over an unrecorded object +// and releases the caller's hold on it when that reader closes, or right away +// if it can't be opened. +func (p *Proxy) serveUnrecorded(ctx context.Context, u *unrecordedObject, artifact artifacts.Artifact, path string) (*CacheResult, error) { + res, err := p.openStoredArtifact(ctx, artifact, path) + if err != nil { + u.release() + return nil, err + } + r := releaseOnClose{ReadCloser: res.Reader, release: u.release} + if s, ok := res.Reader.(io.Seeker); ok { + res.Reader = &seekableReleaseOnClose{releaseOnClose: r, Seeker: s} + } else { + res.Reader = &r + } + return res, nil +} + // openStoredArtifact gives one caller its own reader over just-committed // bytes. Callers sharing a fetch cannot share a handle: it has one read // position, so they would consume each other's bytes and the first Close would @@ -676,13 +742,17 @@ func (p *Proxy) cachedArtifactRecord(pkgPURL, versionPURL, filename, upstreamHas // fetch panicked out of it. var errSharedFetchAbandoned = errors.New("shared upstream fetch did not complete") -// inflightFetch is one upstream fetch that concurrent callers share. val and -// err are written before done closes and read only after, so the close is the -// handoff. +// inflightFetch is one upstream fetch that concurrent callers share. val, err +// and unrecorded are written before done closes and read only after, so the +// close is the handoff. waiters counts callers waiting on it, guarded by +// fetchMu; done closes under fetchMu too, so a waiter leaving early knows +// whether it was counted. type inflightFetch struct { - done chan struct{} - val fetchedArtifact - err error + done chan struct{} + val fetchedArtifact + err error + unrecorded *unrecordedObject + waiters int } // coalesceFetch runs commit at most once for concurrent callers sharing key, @@ -715,7 +785,9 @@ func (p *Proxy) coalesceFetch(ctx context.Context, key string, recheck func() (a p.inFlight = make(map[string]*inflightFetch) } f, joined := p.inFlight[key] - if !joined { + if joined { + f.waiters++ + } else { f = &inflightFetch{done: make(chan struct{})} p.inFlight[key] = f } @@ -728,15 +800,35 @@ func (p *Proxy) coalesceFetch(ctx context.Context, key string, recheck func() (a select { case <-ctx.Done(): // This caller gave up; the fetch continues for everyone else. + p.leaveFetch(f) return nil, ctx.Err() case <-f.done: } if f.err != nil { return nil, f.err } + if f.unrecorded != nil { + return p.serveUnrecorded(ctx, f.unrecorded, f.val.artifact, f.val.storagePath) + } return p.openStoredArtifact(ctx, f.val.artifact, f.val.storagePath) } +// leaveFetch withdraws a waiter that gave up. Once the fetch has finished the +// waiter was counted as holding any unrecorded object, so it releases it. +func (p *Proxy) leaveFetch(f *inflightFetch) { + p.fetchMu.Lock() + select { + case <-f.done: + p.fetchMu.Unlock() + if f.unrecorded != nil { + f.unrecorded.release() + } + default: + f.waiters-- + p.fetchMu.Unlock() + } +} + // runSharedFetch performs the fetch that joined callers are waiting on. It is // never abandoned early, and always releases the key and wakes the waiters. func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch, recheck func() (artifacts.Artifact, string, bool), commit func(context.Context) (artifacts.Artifact, string, error)) (*CacheResult, error) { @@ -746,8 +838,14 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch defer func() { p.fetchMu.Lock() delete(p.inFlight, key) - p.fetchMu.Unlock() + if f.unrecorded != nil { + // Hand each waiter a hold, then drop the one kept while the + // waiters were still unknown. + f.unrecorded.refs.Add(int32(f.waiters)) + defer f.unrecorded.release() + } close(f.done) + p.fetchMu.Unlock() }() // A caller checks the cache before reaching here, so a fetch that finished @@ -767,13 +865,29 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch commitCtx = context.WithoutCancel(ctx) } stored, path, err := commit(commitCtx) + unrecorded := errors.Is(err, errUnrecorded) + if unrecorded { + err = nil + } f.val, f.err = fetchedArtifact{artifact: stored, storagePath: path}, err if err != nil { return nil, err } + if unrecorded { + f.unrecorded = &unrecordedObject{delete: func() { p.deleteUnrecorded(path) }} + f.unrecorded.refs.Add(1) // this caller + f.unrecorded.refs.Add(1) // until the waiters are counted + return p.serveUnrecorded(ctx, f.unrecorded, stored, path) + } return p.openStoredArtifact(ctx, stored, path) } +func (p *Proxy) deleteUnrecorded(path string) { + if err := p.Storage.Delete(context.Background(), path); err != nil { + p.Logger.Warn("failed to delete unrecorded artifact", "path", path, "error", err) + } +} + type detachUpstreamFetchKey struct{} // DetachUpstreamFetch marks ctx so that an artifact fetch started under it diff --git a/internal/handler/unrecorded_test.go b/internal/handler/unrecorded_test.go new file mode 100644 index 00000000..87b6a26a --- /dev/null +++ b/internal/handler/unrecorded_test.go @@ -0,0 +1,304 @@ +package handler + +import ( + "context" + "errors" + "io" + "maps" + "net/http" + "net/http/httptest" + "slices" + "sync" + "testing" + "time" +) + +// These tests refetch an artifact whose stored file is gone while +// pending_deletes refuses writes. The record can't move to the new file and +// the file can't be queued, so the fetch must serve it to every caller sharing +// it and delete it once the last of them is done. + +const ( + unrecordedContent = "refetched content" + unrecordedURL = "https://example.com/gone.tgz" +) + +func setupUnrecorded(t *testing.T) (*Proxy, *mockStorage) { + t.Helper() + proxy, db, store, _ := setupTestProxy(t) + seedPackage(t, db, store, "npm", "gone", "1.0.0", "gone-1.0.0.tgz", "old content") + oldPath := recordedStoragePath(t, db, "pkg:npm/gone@1.0.0", "gone-1.0.0.tgz") + delete(store.files, oldPath) + if _, err := db.Exec(`CREATE TRIGGER refuse_queue BEFORE INSERT ON pending_deletes + BEGIN SELECT RAISE(ABORT, 'queue refused'); END`); err != nil { + t.Fatalf("creating trigger: %v", err) + } + t.Cleanup(func() { + if got := recordedStoragePath(t, db, "pkg:npm/gone@1.0.0", "gone-1.0.0.tgz"); got != oldPath { + t.Errorf("record points at %q, want it kept at %q", got, oldPath) + } + }) + return proxy, store +} + +func fetchGone(ctx context.Context, proxy *Proxy) (*CacheResult, error) { + return proxy.GetOrFetchArtifactFromURL(ctx, "npm", "gone", "1.0.0", "gone-1.0.0.tgz", unrecordedURL) +} + +func storedFiles(store *mockStorage) []string { + store.mu.Lock() + defer store.mu.Unlock() + return slices.Collect(maps.Keys(store.files)) +} + +func readAndClose(t *testing.T, res *CacheResult) { + t.Helper() + body, err := io.ReadAll(res.Reader) + _ = res.Reader.Close() + if err != nil || string(body) != unrecordedContent { + t.Errorf("body = %q, %v, want %q", body, err, unrecordedContent) + } +} + +// waitForWaiters blocks until n callers wait on the one fetch in flight. +func waitForWaiters(t *testing.T, proxy *Proxy, n int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + proxy.fetchMu.Lock() + got := -1 + for _, f := range proxy.inFlight { + got = f.waiters + } + proxy.fetchMu.Unlock() + if got == n { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("never saw %d waiters on the fetch", n) +} + +func TestUnrecordedRefetchIsDeletedOnceServed(t *testing.T) { + proxy, store := setupUnrecorded(t) + fetcher := &countingFetcher{content: unrecordedContent} + proxy.Fetcher = fetcher + + for range 2 { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Fatalf("fetch: %v", err) + } + if got := storedFiles(store); len(got) != 1 { + t.Errorf("stored files while serving = %v, want the refetched one", got) + } + readAndClose(t, res) + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after serving = %v, want none left untracked", got) + } + } +} + +func TestUnrecordedRefetchServesRangeRequests(t *testing.T) { + proxy, store := setupUnrecorded(t) + store.seekable = true + proxy.Fetcher = &countingFetcher{content: unrecordedContent} + + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Fatalf("fetch: %v", err) + } + req := httptest.NewRequest(http.MethodGet, "/gone-1.0.0.tgz", nil) + req.Header.Set("Range", "bytes=2-5") + rec := httptest.NewRecorder() + ServeArtifactRequest(rec, req, res) + + if rec.Code != http.StatusPartialContent || rec.Body.String() != unrecordedContent[2:6] { + t.Errorf("range response = %d %q, want 206 %q", rec.Code, rec.Body.String(), unrecordedContent[2:6]) + } + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after serving = %v, want none left untracked", got) + } +} + +// TestUnrecordedRefetchWaitsForEveryCaller keeps the object until the caller +// that ran the fetch and every caller that waited on it have closed. +func TestUnrecordedRefetchWaitsForEveryCaller(t *testing.T) { + const waiters = 4 + proxy, store := setupUnrecorded(t) + fetcher := newGatedFetcher(unrecordedContent) + proxy.Fetcher = fetcher + + leader := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("leader: %v", err) + } + leader <- res + }() + <-fetcher.entered + + results := make(chan *CacheResult, waiters) + var wg sync.WaitGroup + for range waiters { + wg.Go(func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("waiter: %v", err) + return + } + results <- res + }) + } + waitForWaiters(t, proxy, waiters) + close(fetcher.release) + + // The leader holds its reader open while every waiter reads and closes. + lead := <-leader + wg.Wait() + close(results) + for res := range results { + readAndClose(t, res) + } + if got := storedFiles(store); len(got) != 1 { + t.Fatalf("stored files with the leader still reading = %v, want the refetched one", got) + } + readAndClose(t, lead) + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after every caller closed = %v, want none", got) + } +} + +// ctxStorage fails to open under a cancelled context, as cloud backends do. +type ctxStorage struct{ *mockStorage } + +func (s ctxStorage) Open(ctx context.Context, path string) (io.ReadCloser, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return s.mockStorage.Open(ctx, path) +} + +// TestUnrecordedRefetchOutlivesDetachedCaller is the retry a detached fetch +// exists for: the client that started it gave up, and its retry waits on the +// same fetch. The retry must get the bytes rather than a deleted object. +func TestUnrecordedRefetchOutlivesDetachedCaller(t *testing.T) { + proxy, store := setupUnrecorded(t) + proxy.Storage = ctxStorage{store} + fetcher := newGatedFetcher(unrecordedContent) + proxy.Fetcher = fetcher + + ctx, cancel := context.WithCancel(DetachUpstreamFetch(context.Background())) + leaderDone := make(chan struct{}) + go func() { + defer close(leaderDone) + res, _ := fetchGone(ctx, proxy) + drain(res) + }() + <-fetcher.entered + cancel() + + retry := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("retry: %v", err) + } + retry <- res + }() + waitForWaiters(t, proxy, 1) + close(fetcher.release) + <-leaderDone + + if res := <-retry; res != nil { + readAndClose(t, res) + } + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after the retry closed = %v, want none", got) + } +} + +// TestUnrecordedRefetchIgnoresWaiterThatLeft checks that a waiter giving up +// before the fetch ends does not keep the object forever. +func TestUnrecordedRefetchIgnoresWaiterThatLeft(t *testing.T) { + proxy, store := setupUnrecorded(t) + fetcher := newGatedFetcher(unrecordedContent) + proxy.Fetcher = fetcher + + leader := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("leader: %v", err) + } + leader <- res + }() + <-fetcher.entered + + ctx, cancel := context.WithCancel(context.Background()) + left := make(chan error, 1) + go func() { + _, err := fetchGone(ctx, proxy) + left <- err + }() + waitForWaiters(t, proxy, 1) + cancel() + if err := <-left; !errors.Is(err, context.Canceled) { + t.Fatalf("waiter that left: err = %v, want context.Canceled", err) + } + close(fetcher.release) + + readAndClose(t, <-leader) + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after the leader closed = %v, want none", got) + } +} + +// TestLeaveFetch covers both ways a waiter can give up: before the fetch +// finished, when it was not handed a hold yet, and after, when it was. +func TestLeaveFetch(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + deleted := false + newFetch := func() *inflightFetch { + f := &inflightFetch{done: make(chan struct{}), waiters: 1} + f.unrecorded = &unrecordedObject{delete: func() { deleted = true }} + f.unrecorded.refs.Add(1) + return f + } + + f := newFetch() + proxy.leaveFetch(f) + if f.waiters != 0 || deleted { + t.Errorf("before done: waiters = %d, deleted = %v, want 0 and false", f.waiters, deleted) + } + + f = newFetch() + close(f.done) + proxy.leaveFetch(f) + if f.waiters != 1 || !deleted { + t.Errorf("after done: waiters = %d, deleted = %v, want 1 and true", f.waiters, deleted) + } +} + +// TestRecordedAt is the check that keeps an object whose record update +// committed despite reporting an error. +func TestRecordedAt(t *testing.T) { + proxy, db, store, _ := setupTestProxy(t) + seedPackage(t, db, store, "npm", "kept", "1.0.0", "kept-1.0.0.tgz", "content") + path := recordedStoragePath(t, db, "pkg:npm/kept@1.0.0", "kept-1.0.0.tgz") + + if !proxy.recordedAt("pkg:npm/kept@1.0.0", "kept-1.0.0.tgz", path) { + t.Error("record pointing at the path not seen") + } + if proxy.recordedAt("pkg:npm/kept@1.0.0", "kept-1.0.0.tgz", "other/path.tgz") { + t.Error("record pointing elsewhere taken as recorded") + } + if proxy.recordedAt("pkg:npm/none@1.0.0", "none-1.0.0.tgz", path) { + t.Error("missing record taken as recorded") + } + _ = db.Close() + if proxy.recordedAt("pkg:npm/kept@1.0.0", "kept-1.0.0.tgz", path) { + t.Error("failed lookup taken as recorded") + } +} From 257d159ecf22633da5ffd2e121907d663b0cba84 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Mon, 26 Oct 2026 10:19:24 +0000 Subject: [PATCH 3/3] Release an unrecorded object's hold only once per reader --- internal/handler/handler.go | 3 ++- internal/handler/unrecorded_test.go | 38 +++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/internal/handler/handler.go b/internal/handler/handler.go index a4007fcb..fc5b912f 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -802,7 +802,8 @@ func (p *Proxy) serveUnrecorded(ctx context.Context, u *unrecordedObject, artifa u.release() return nil, err } - r := releaseOnClose{ReadCloser: res.Reader, release: u.release} + // Once, so a second Close can't drop another caller's hold. + r := releaseOnClose{ReadCloser: res.Reader, release: sync.OnceFunc(u.release)} if s, ok := res.Reader.(io.Seeker); ok { res.Reader = &seekableReleaseOnClose{releaseOnClose: r, Seeker: s} } else { diff --git a/internal/handler/unrecorded_test.go b/internal/handler/unrecorded_test.go index 87b6a26a..3533a10e 100644 --- a/internal/handler/unrecorded_test.go +++ b/internal/handler/unrecorded_test.go @@ -170,6 +170,44 @@ func TestUnrecordedRefetchWaitsForEveryCaller(t *testing.T) { } } +func TestUnrecordedRefetchSurvivesDoubleClose(t *testing.T) { + proxy, store := setupUnrecorded(t) + fetcher := newGatedFetcher(unrecordedContent) + proxy.Fetcher = fetcher + + leader := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("leader: %v", err) + } + leader <- res + }() + <-fetcher.entered + + waiter := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("waiter: %v", err) + } + waiter <- res + }() + waitForWaiters(t, proxy, 1) + close(fetcher.release) + + lead, wait := <-leader, <-waiter + readAndClose(t, lead) + _ = lead.Reader.Close() + if got := storedFiles(store); len(got) != 1 { + t.Fatalf("stored files with the waiter still reading = %v, want the refetched one", got) + } + readAndClose(t, wait) + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after every caller closed = %v, want none", got) + } +} + // ctxStorage fails to open under a cancelled context, as cloud backends do. type ctxStorage struct{ *mockStorage }