Clear a cached artifact and queue its file in one transaction

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"} {

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.

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