From e6375011dccd54c01fed171af1e70166451f2428 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 25 Oct 2026 13:37:03 +1100 Subject: [PATCH] Harden github_rpm repodata cache error, 404 and store ordering --- .../api/v2/remotes_cache_repodata_test.go | 57 ++++++ internal/provider/rpm/github.go | 46 +++-- .../rpm/github_repodata_cache_test.go | 179 +++++++++++++++++- internal/provider/rpm/rpm.go | 20 +- 4 files changed, 273 insertions(+), 29 deletions(-) create mode 100644 internal/api/v2/remotes_cache_repodata_test.go diff --git a/internal/api/v2/remotes_cache_repodata_test.go b/internal/api/v2/remotes_cache_repodata_test.go new file mode 100644 index 0000000..6e9d23a --- /dev/null +++ b/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) + } +} diff --git a/internal/provider/rpm/github.go b/internal/provider/rpm/github.go index d534e17..e459690 100644 --- a/internal/provider/rpm/github.go +++ b/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) diff --git a/internal/provider/rpm/github_repodata_cache_test.go b/internal/provider/rpm/github_repodata_cache_test.go index 5d19382..6cd5388 100644 --- a/internal/provider/rpm/github_repodata_cache_test.go +++ b/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()) + } +} diff --git a/internal/provider/rpm/rpm.go b/internal/provider/rpm/rpm.go index 723edfa..5e18fcb 100644 --- a/internal/provider/rpm/rpm.go +++ b/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 {