Harden github_rpm repodata cache error, 404 and store ordering

internal/api/v2/remotes_cache_repodata_test.go

@@ -0,0 +1,57 @@
+package v2
+
+import (
+	"encoding/json"
+	"net/http/httptest"
+	"testing"
+
+	"git.unkin.net/unkin/artifactapi/internal/database"
+)
+
+func TestCacheRepodataRoundTrip(t *testing.T) {
+	if testDSN == "" {
+		t.Skip("Docker unavailable")
+	}
+	db, err := database.New(testDSN)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	h := NewRemotesHandler(db, nil, nil).Routes()
+	get := func() bool {
+		t.Helper()
+		w := httptest.NewRecorder()
+		h.ServeHTTP(w, httptest.NewRequest("GET", "/cache-rd", nil))
+		var got struct {
+			CacheRepodata *bool `json:"cache_repodata"`
+		}
+		if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got.CacheRepodata == nil {
+			t.Fatalf("get = %d %s", w.Code, w.Body.String())
+		}
+		return *got.CacheRepodata
+	}
+	const body = `"package_type":"github_rpm","base_url":"https://api.github.com"`
+
+	steps := []struct {
+		method, path, json string
+		want               bool
+	}{
+		{"POST", "/", `{"name":"cache-rd",` + body + `}`, false},
+		{"PUT", "/cache-rd", `{` + body + `,"cache_repodata":true}`, true},
+		{"PUT", "/cache-rd", `{` + body + `,"cache_repodata":false}`, false},
+	}
+	for i, s := range steps {
+		if c := do(t, h, s.method, s.path, s.json); c/100 != 2 {
+			t.Fatalf("step %d %s %s = %d", i, s.method, s.path, c)
+		}
+		if got := get(); got != s.want {
+			t.Errorf("step %d: cache_repodata = %v, want %v", i, got, s.want)
+		}
+	}
+	if c := do(t, h, "POST", "/", `{"name":"cache-rd-on",`+body+`,"cache_repodata":true}`); c != 201 {
+		t.Fatalf("create with cache_repodata = %d", c)
+	}
+	if r, err := db.GetRemote(t.Context(), "cache-rd-on"); err != nil || !r.CacheRepodata {
+		t.Fatalf("create did not store cache_repodata: %v %v", r, err)
+	}
+}

internal/provider/rpm/github.go

@@ -15,6 +15,7 @@ import (
 	"regexp"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	rpmlib "github.com/cavaliergopher/rpm"
@@ -54,7 +55,8 @@ const (
 	// of a large repo; small repos usually prime within this window.
 	defaultColdWait = 8 * time.Second
 	// repodataIdleTTL evicts cached indexes of remotes deleted or switched off.
-	repodataIdleTTL = time.Hour
+	repodataIdleTTL    = time.Hour
+	repodataSweepEvery = time.Minute
 )
 
 // GitHubProvider is a metadata-only remote: it scans a GitHub repo's releases
@@ -92,6 +94,8 @@ type GitHubProvider struct {
 	repodataMu    sync.Mutex
 	repodataCache map[string]cachedRepodata
 	repodataGroup singleflight.Group
+	repodataSeq   atomic.Int64
+	repodataSwept time.Time
 }
 
 // rpmFingerprinter is a cheap change detector over a remote's metadata rows.
@@ -102,6 +106,7 @@ type rpmFingerprinter interface {
 
 type cachedRepodata struct {
 	count, maxID int64
+	seq          int64 // build order, so a slow older build never replaces a newer one
 	used         time.Time
 	*repodata
 }
@@ -270,30 +275,36 @@ func (p *GitHubProvider) ensurePrimed(ctx context.Context, remote models.Remote,
 // database on every request, so each replica rebuilds as soon as any replica's
 // scan changes rows and repomd never outlives the files it references.
 func (p *GitHubProvider) serveCachedRepodata(w http.ResponseWriter, r *http.Request, store provider.RPMMetadataReader, fp rpmFingerprinter, name, tail string) {
-	rd, err := p.cachedRepodata(r.Context(), store, fp, name)
-	if err != nil {
-		slog.Error("github_rpm: build repodata failed", "remote", name, "error", err)
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
-	body, contentType := rd.other, "application/gzip"
+	var pick func(*repodata) []byte
+	contentType := "application/gzip"
 	switch {
 	case tail == "repomd.xml":
-		body, contentType = rd.repomd, "application/xml"
+		pick, contentType = func(rd *repodata) []byte { return rd.repomd }, "application/xml"
 	case strings.HasSuffix(tail, "-primary.xml.gz"):
-		body = rd.primary
+		pick = func(rd *repodata) []byte { return rd.primary }
 	case strings.HasSuffix(tail, "-filelists.xml.gz"):
-		body = rd.filelists
-	case !strings.HasSuffix(tail, "-other.xml.gz"):
+		pick = func(rd *repodata) []byte { return rd.filelists }
+	case strings.HasSuffix(tail, "-other.xml.gz"):
+		pick = func(rd *repodata) []byte { return rd.other }
+	default:
 		http.Error(w, "not found", http.StatusNotFound)
 		return
 	}
+	rd, err := p.cachedRepodata(r.Context(), store, fp, name)
+	if err != nil {
+		writeMetadataError(w, name, err)
+		return
+	}
 	w.Header().Set("Content-Type", contentType)
 	w.WriteHeader(http.StatusOK)
-	_, _ = w.Write(body)
+	_, _ = w.Write(pick(rd))
 }
 
+// cachedRepodata returns the index for the current fingerprint, building it at
+// most once per fingerprint. ctx is ServeRemote's detached serve context, so a
+// leader's client disconnect cannot fail the build for coalesced waiters.
 func (p *GitHubProvider) cachedRepodata(ctx context.Context, store provider.RPMMetadataReader, fp rpmFingerprinter, name string) (*repodata, error) {
+	seq := p.repodataSeq.Add(1)
 	count, maxID, err := fp.RPMMetadataFingerprint(ctx, name)
 	if err != nil {
 		return nil, err
@@ -320,7 +331,9 @@ func (p *GitHubProvider) cachedRepodata(ctx context.Context, store provider.RPMM
 		// worst case is one extra rebuild on the next request.
 		rd := buildRepodata(metas)
 		p.repodataMu.Lock()
-		p.repodataCache[name] = cachedRepodata{count: count, maxID: maxID, used: time.Now(), repodata: rd}
+		if cur, ok := p.repodataCache[name]; !ok || cur.seq < seq {
+			p.repodataCache[name] = cachedRepodata{count: count, maxID: maxID, seq: seq, used: time.Now(), repodata: rd}
+		}
 		p.repodataMu.Unlock()
 		return rd, nil
 	})
@@ -337,7 +350,12 @@ func (p *GitHubProvider) dropRepodata(name string) {
 	p.repodataMu.Unlock()
 }
 
+// sweepRepodataLocked evicts idle entries, at most once per repodataSweepEvery.
 func (p *GitHubProvider) sweepRepodataLocked(now time.Time) {
+	if now.Sub(p.repodataSwept) < repodataSweepEvery {
+		return
+	}
+	p.repodataSwept = now
 	for name, c := range p.repodataCache {
 		if now.Sub(c.used) > repodataIdleTTL {
 			delete(p.repodataCache, name)

internal/provider/rpm/github_repodata_cache_test.go

@@ -2,9 +2,11 @@ package rpm
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -15,14 +17,23 @@ import (
 // fpStore adds a fingerprint and a full-read counter to fakeStore.
 type fpStore struct {
 	*fakeStore
-	lists int
-	seq   int64
+	lists, fps     int
+	seq            int64
+	fpErr, listErr error
+	gate           chan struct{} // when set, full reads block until it is closed
 }
 
 func (s *fpStore) ListRPMMetadataEntries(ctx context.Context, name string) ([]provider.RPMMetadata, error) {
 	s.mu.Lock()
 	s.lists++
+	gate, err := s.gate, s.listErr
 	s.mu.Unlock()
+	if gate != nil {
+		<-gate
+	}
+	if err != nil {
+		return nil, err
+	}
 	return s.fakeStore.ListRPMMetadataEntries(ctx, name)
 }
 
@@ -42,7 +53,14 @@ func (s *fpStore) InsertRPMMetadata(ctx context.Context, m *provider.RPMMetadata
 func (s *fpStore) RPMMetadataFingerprint(_ context.Context, _ string) (int64, int64, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return int64(len(s.rows)), s.seq, nil
+	s.fps++
+	return int64(len(s.rows)), s.seq, s.fpErr
+}
+
+func (s *fpStore) fpCount() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.fps
 }
 
 func (s *fpStore) listCount() int {
@@ -72,13 +90,18 @@ func addPkg(t *testing.T, store *fpStore, name string) {
 	}
 }
 
-func serve(t *testing.T, p *GitHubProvider, remote models.Remote, store *fpStore, path string) []byte {
+func serveReq(t *testing.T, p *GitHubProvider, remote models.Remote, store *fpStore, req *http.Request, path string) *httptest.ResponseRecorder {
 	t.Helper()
 	rec := httptest.NewRecorder()
-	req := httptest.NewRequest(http.MethodGet, "/x", nil)
 	if !p.ServeRemote(rec, req, remote, path, "https://proxy.example", store) {
-		t.Fatalf("%s not handled", path)
+		t.Errorf("%s not handled", path)
 	}
+	return rec
+}
+
+func serve(t *testing.T, p *GitHubProvider, remote models.Remote, store *fpStore, path string) []byte {
+	t.Helper()
+	rec := serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), path)
 	if rec.Code != http.StatusOK {
 		t.Fatalf("%s: code %d: %s", path, rec.Code, rec.Body.String())
 	}
@@ -97,9 +120,6 @@ func TestCachedRepodataHitSkipsRegeneration(t *testing.T) {
 
 	repomd := serve(t, p, remote, store, "repodata/repomd.xml")
 	primary := serve(t, p, remote, store, "repodata/x-primary.xml.gz")
-	for _, path := range repodataPaths {
-		serve(t, p, remote, store, path)
-	}
 	if n := store.listCount(); n != 1 {
 		t.Fatalf("full metadata reads = %d, want 1", n)
 	}
@@ -175,6 +195,7 @@ func TestCachedRepodataEvictsIdleRemotes(t *testing.T) {
 	p, store, remote := newCacheFixture(t, true)
 	serve(t, p, remote, store, "repodata/repomd.xml")
 	p.repodataCache["gone"] = cachedRepodata{used: time.Now().Add(-2 * repodataIdleTTL), repodata: &repodata{}}
+	p.repodataSwept = time.Time{}
 
 	serve(t, p, remote, store, "repodata/repomd.xml")
 	if _, ok := p.repodataCache["gone"]; ok {
@@ -184,3 +205,143 @@ func TestCachedRepodataEvictsIdleRemotes(t *testing.T) {
 		t.Fatal("active remote's index evicted")
 	}
 }
+
+// waitFor polls until cond holds, then gives goroutines that passed the
+// fingerprint a moment to reach the singleflight call.
+func waitFor(t *testing.T, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatal("condition not reached")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	time.Sleep(50 * time.Millisecond)
+}
+
+func TestCachedRepodataCoalescesConcurrentBuilds(t *testing.T) {
+	p, store, remote := newCacheFixture(t, true)
+	gate := make(chan struct{})
+	store.gate = gate
+	const n = 8
+	var wg sync.WaitGroup
+	codes := make([]int, n)
+	for i := range n {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			codes[i] = serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), repodataPaths[i%len(repodataPaths)]).Code
+		}()
+	}
+	waitFor(t, func() bool { return store.fpCount() == n })
+	close(gate)
+	wg.Wait()
+	for i, c := range codes {
+		if c != http.StatusOK {
+			t.Errorf("request %d: code %d", i, c)
+		}
+	}
+	if got := store.listCount(); got != 1 {
+		t.Fatalf("full metadata reads = %d, want 1", got)
+	}
+}
+
+func TestCachedRepodataLeaderCancelDoesNotFailWaiters(t *testing.T) {
+	p, store, remote := newCacheFixture(t, true)
+	gate := make(chan struct{})
+	store.gate = gate
+	leaderCtx, cancelLeader := context.WithCancel(context.Background())
+	results := make(chan int, 2)
+	go func() {
+		req := httptest.NewRequest(http.MethodGet, "/x", nil).WithContext(leaderCtx)
+		results <- serveReq(t, p, remote, store, req, "repodata/repomd.xml").Code
+	}()
+	waitFor(t, func() bool { return store.listCount() == 1 })
+	go func() {
+		results <- serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), "repodata/repomd.xml").Code
+	}()
+	waitFor(t, func() bool { return store.fpCount() == 2 })
+	cancelLeader()
+	close(gate)
+	for range 2 {
+		if c := <-results; c != http.StatusOK {
+			t.Fatalf("code %d after leader cancel, want 200", c)
+		}
+	}
+	if got := store.listCount(); got != 1 {
+		t.Fatalf("full metadata reads = %d, want 1", got)
+	}
+}
+
+func TestCachedRepodataOlderBuildDoesNotOverwriteNewer(t *testing.T) {
+	p, store, remote := newCacheFixture(t, true)
+	newer := &repodata{}
+	p.repodataCache[remote.Name] = cachedRepodata{count: 99, maxID: 99, seq: 1 << 40, used: time.Now(), repodata: newer}
+	serve(t, p, remote, store, "repodata/repomd.xml")
+	if p.repodataCache[remote.Name].repodata != newer {
+		t.Fatal("older build replaced a newer cache entry")
+	}
+}
+
+func TestCachedRepodataErrors(t *testing.T) {
+	secret := errors.New("pq: password authentication failed for user secret")
+	for _, tc := range []struct {
+		name           string
+		fpErr, listErr error
+		code           int
+	}{
+		{"fingerprint", secret, nil, http.StatusInternalServerError},
+		{"list", nil, secret, http.StatusInternalServerError},
+		{"fingerprint deadline", context.DeadlineExceeded, nil, http.StatusServiceUnavailable},
+		{"list canceled", nil, context.Canceled, http.StatusServiceUnavailable},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			p, store, remote := newCacheFixture(t, true)
+			store.fpErr, store.listErr = tc.fpErr, tc.listErr
+			rec := serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), "repodata/repomd.xml")
+			if rec.Code != tc.code {
+				t.Fatalf("code %d, want %d", rec.Code, tc.code)
+			}
+			if strings.Contains(rec.Body.String(), "secret") {
+				t.Fatalf("DB error leaked: %q", rec.Body.String())
+			}
+			if retry := rec.Header().Get("Retry-After"); (tc.code == http.StatusServiceUnavailable) != (retry != "") {
+				t.Fatalf("Retry-After = %q for %d", retry, tc.code)
+			}
+			if _, ok := p.repodataCache[remote.Name]; ok {
+				t.Fatal("failed build was cached")
+			}
+		})
+	}
+}
+
+func TestCachedRepodataUnknownTail404sWithoutDB(t *testing.T) {
+	p, store, remote := newCacheFixture(t, true)
+	rec := serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), "repodata/bogus.txt")
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("code %d, want 404", rec.Code)
+	}
+	if store.fpCount() != 0 || store.listCount() != 0 {
+		t.Fatalf("unknown tail hit the DB: fps=%d lists=%d", store.fpCount(), store.listCount())
+	}
+}
+
+func TestCacheEmptyUsesFingerprint(t *testing.T) {
+	p := newGitHubProvider()
+	store := &fpStore{fakeStore: newFakeStore()}
+	if !p.cacheEmpty(context.Background(), store, "cached-rpm") {
+		t.Fatal("empty store not reported empty")
+	}
+	addPkg(t, store, "alpha")
+	if p.cacheEmpty(context.Background(), store, "cached-rpm") {
+		t.Fatal("populated store reported empty")
+	}
+	store.fpErr = errors.New("db down")
+	if p.cacheEmpty(context.Background(), store, "cached-rpm") {
+		t.Fatal("fingerprint error must not report empty")
+	}
+	if store.listCount() != 0 {
+		t.Fatalf("cacheEmpty did full reads: %d", store.listCount())
+	}
+}

internal/provider/rpm/rpm.go

@@ -250,17 +250,25 @@ func (p *Provider) GenerateLocalIndex(ctx context.Context, files provider.FileSt
 func readMetadataEntries(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) ([]provider.RPMMetadata, bool) {
 	metas, err := reader.ListRPMMetadataEntries(r.Context(), repoName)
 	if err != nil {
-		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
-			slog.Warn("rpm: metadata read canceled", "repo", repoName, "error", err)
-			http.Error(w, "metadata read canceled", http.StatusServiceUnavailable)
-			return nil, false
-		}
-		http.Error(w, err.Error(), http.StatusInternalServerError)
+		writeMetadataError(w, repoName, err)
 		return nil, false
 	}
 	return metas, true
 }
 
+// writeMetadataError maps a metadata read failure to a response without leaking
+// DB error text: a canceled/expired context is a retryable 503, anything else a 500.
+func writeMetadataError(w http.ResponseWriter, repoName string, err error) {
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		slog.Warn("rpm: metadata read canceled", "repo", repoName, "error", err)
+		w.Header().Set("Retry-After", "5")
+		http.Error(w, "metadata read canceled", http.StatusServiceUnavailable)
+		return
+	}
+	slog.Error("rpm: metadata read failed", "repo", repoName, "error", err)
+	http.Error(w, "metadata read failed", http.StatusInternalServerError)
+}
+
 func (p *Provider) serveRepomd(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) {
 	metas, ok := readMetadataEntries(w, r, reader, repoName)
 	if !ok {