diff --git a/internal/database/pending_deletes_test.go b/internal/database/pending_deletes_test.go index 82bdacaa..f1245571 100644 --- a/internal/database/pending_deletes_test.go +++ b/internal/database/pending_deletes_test.go @@ -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 7af5d070..d7150e80 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -581,33 +581,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/handler/handler.go b/internal/handler/handler.go index 363314e8..fc5b912f 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" @@ -737,14 +738,80 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil } // 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 + } + // 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 { + 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 @@ -796,15 +863,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. done closes under fetchMu, so a waiter leaving early +// knows whether it was counted in waiters. type inflightFetch struct { - done chan struct{} - val fetchedArtifact - err error + done chan struct{} + val fetchedArtifact + err error + unrecorded *unrecordedObject - // waiters counts callers that joined the fetch, guarded by fetchMu. + // waiters counts callers waiting on the fetch, guarded by fetchMu. // Tests read it to know a caller has joined. waiters int } @@ -854,15 +923,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) { @@ -872,8 +961,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 @@ -935,14 +1030,32 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch commitCtx = context.WithoutCancel(ctx) } stored, path, err := commit(commitCtx) - failed = err != nil + unrecorded := errors.Is(err, errUnrecorded) + if unrecorded { + err = nil + } + // Nothing an instance waiting on the lease could find was recorded, so + // an unrecorded object counts as failed for it. + failed = err != nil || unrecorded 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 servingHeadKey struct{} // ServingHead marks ctx as serving a HEAD request. A cache hit under it is diff --git a/internal/handler/unrecorded_test.go b/internal/handler/unrecorded_test.go new file mode 100644 index 00000000..3533a10e --- /dev/null +++ b/internal/handler/unrecorded_test.go @@ -0,0 +1,342 @@ +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) + } +} + +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 } + +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") + } +} diff --git a/internal/server/eviction.go b/internal/server/eviction.go index 66887d18..5e066bde 100644 --- a/internal/server/eviction.go +++ b/internal/server/eviction.go @@ -125,6 +125,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 := clearEntry(db, logger, entry) if err != nil { logger.Warn("eviction: failed to clear record", @@ -138,15 +139,9 @@ func evictBatch(ctx context.Context, db *database.DB, store storage.Storage, log // Metadata goes now too: readers hold the whole document in memory, // and a row refetched meanwhile refetches again on a missing blob. 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) - } - } else if entry.Kind == database.CacheEntryMetadata { - // Queued with the row's delete in case the process stopped here. - if err := db.RemovePendingDelete(path); err != nil { - logger.Warn("eviction: failed to dequeue deleted object", "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 entry.Size.Valid { diff --git a/internal/server/eviction_test.go b/internal/server/eviction_test.go index 0567499f..99977fe6 100644 --- a/internal/server/eviction_test.go +++ b/internal/server/eviction_test.go @@ -488,3 +488,67 @@ 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, artifactEntries(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, artifactEntries(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) + } +} + +// artifactEntries turns LRU artifact rows into eviction entries. +func artifactEntries(rows []database.Artifact) []database.CacheEntry { + entries := make([]database.CacheEntry, len(rows)) + for i, a := range rows { + entries[i] = database.CacheEntry{Kind: database.CacheEntryArtifact, Key: a.VersionPURL, Name: a.Filename, StoragePath: a.StoragePath, Size: a.Size} + } + return entries +}