diff --git a/docs/rpm/remote-github.md b/docs/rpm/remote-github.md index ccb9e69..6180e8a 100644 --- a/docs/rpm/remote-github.md +++ b/docs/rpm/remote-github.md @@ -72,6 +72,7 @@ resource "artifactapi_remote_github_rpm" "acme-tools" { base_url = "https://api.github.com/repos/acme/tools" # the releases API root releases_remote = "github" # backend for downloads mutable_ttl = 3600 # release re-scan interval + cache_repodata = true # keep generated repodata in memory # Optional: restrict which release assets become packages (regex on filename). patterns = [".*\\.x86_64\\.rpm$", ".*\\.noarch\\.rpm$"] @@ -84,3 +85,11 @@ resource "artifactapi_remote_github_rpm" "acme-tools" { `dnf` config: `baseurl=https://artifactapi.example/api/v1/remote/acme-tools`. The repo is multi-arch (no `$basearch` needed) — `dnf` selects matching packages from the synthesized metadata. + +### Repodata cache + +Repodata is regenerated from the metadata rows on every request by default. Set +`cache_repodata = true` on the remote to keep the generated `repomd.xml` and its +`primary`/`filelists`/`other` files in memory per replica. Each request checks a +cheap row fingerprint (count + newest id) in the database and rebuilds only when +it moved, so every replica picks up a scan's changes on its next request. 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/database/database_test.go b/internal/database/database_test.go index c97f314..70a04f1 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -79,13 +79,19 @@ func TestRemotesCRUD(t *testing.T) { if err != nil || got.BaseURL != "https://example.com" { t.Fatalf("get: %v %v", got, err) } + if got.CacheRepodata { + t.Error("cache_repodata should default to false") + } got.BaseURL = "https://updated.example.com" + got.CacheRepodata = true + got.SyncInterval = 600 + got.ServeStale = true if err := testDB.UpdateRemote(ctx(), got); err != nil { t.Fatalf("update: %v", err) } got, _ = testDB.GetRemote(ctx(), "r-crud") - if got.BaseURL != "https://updated.example.com" { - t.Errorf("update not applied: %v", got.BaseURL) + if got.BaseURL != "https://updated.example.com" || !got.CacheRepodata || got.SyncInterval != 600 || !got.ServeStale { + t.Errorf("update not applied: %v %v %v %v", got.BaseURL, got.CacheRepodata, got.SyncInterval, got.ServeStale) } list, err := testDB.ListRemotes(ctx()) if err != nil || len(list) == 0 { @@ -487,3 +493,44 @@ func TestRPMMetadata(t *testing.T) { t.Fatalf("list rpm rows: %v %v", len(rows), err) } } + +func TestRPMMetadataFingerprint(t *testing.T) { + requireDB(t) + seedRemote(t, "r-rpm-fp") + fp := func() [2]int64 { + t.Helper() + c, m, err := testDB.RPMMetadataFingerprint(ctx(), "r-rpm-fp") + if err != nil { + t.Fatal(err) + } + return [2]int64{c, m} + } + insert := func(path string) { + t.Helper() + if err := testDB.InsertRPMMetadata(ctx(), &provider.RPMMetadata{ + RepoName: "r-rpm-fp", FilePath: path, ContentHash: "sha256:" + path, + Name: path, Version: "1", Release: "1", Arch: "noarch", + }); err != nil { + t.Fatal(err) + } + } + + if got := fp(); got != [2]int64{0, 0} { + t.Fatalf("empty fingerprint = %v", got) + } + insert("a.rpm") + one := fp() + insert("a.rpm") + if fp() != one { + t.Fatal("conflicting insert changed fingerprint") + } + insert("b.rpm") + two := fp() + if err := testDB.DeleteRPMMetadata(ctx(), "r-rpm-fp", "a.rpm"); err != nil { + t.Fatal(err) + } + afterDelete := fp() + if one[0] != 1 || two[0] != 2 || afterDelete[0] != 1 || afterDelete == one || afterDelete == two { + t.Fatalf("fingerprints did not track changes: %v %v %v", one, two, afterDelete) + } +} diff --git a/internal/database/remotes.go b/internal/database/remotes.go index d5a70b2..b1a6e44 100644 --- a/internal/database/remotes.go +++ b/internal/database/remotes.go @@ -15,7 +15,8 @@ const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, mirror_ quarantine_enabled, quarantine_days, stale_on_error, releases_remote, managed_by, upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout, - serve_stale, created_at, updated_at` + serve_stale, cache_repodata, + created_at, updated_at` // normalizeMirrorStrategy maps an empty strategy to the round_robin default so // the NOT NULL mirror_strategy column always stores a canonical value. @@ -35,7 +36,8 @@ func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error &r.QuarantineEnabled, &r.QuarantineDays, &r.StaleOnError, &r.ReleasesRemote, &r.ManagedBy, &r.UpstreamDialTimeout, &r.UpstreamTLSTimeout, &r.UpstreamResponseHeaderTimeout, - &r.ServeStale, &r.CreatedAt, &r.UpdatedAt, + &r.ServeStale, &r.CacheRepodata, + &r.CreatedAt, &r.UpdatedAt, ) } @@ -76,8 +78,8 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error { quarantine_enabled, quarantine_days, stale_on_error, releases_remote, managed_by, upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout, - mirror_strategy, sync_interval, serve_stale - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28) + mirror_strategy, sync_interval, serve_stale, cache_repodata + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29) `, r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Mirrorlist, r.Description, r.Username, r.Password, r.ImmutableTTL, r.MutableTTL, r.CheckMutable, @@ -86,7 +88,7 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error { r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError, r.ReleasesRemote, r.ManagedBy, r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout, - normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale, + normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale, r.CacheRepodata, ) return err } @@ -101,7 +103,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error { quarantine_enabled=$17, quarantine_days=$18, stale_on_error=$19, releases_remote=$20, managed_by=$21, upstream_dial_timeout=$22, upstream_tls_timeout=$23, upstream_response_header_timeout=$24, - mirror_strategy=$26, sync_interval=$27, serve_stale=$28, + mirror_strategy=$26, sync_interval=$27, serve_stale=$28, cache_repodata=$29, updated_at=NOW() WHERE name=$1 `, @@ -113,7 +115,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error { r.ReleasesRemote, r.ManagedBy, r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout, r.Mirrorlist, - normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale, + normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale, r.CacheRepodata, ) return err } diff --git a/internal/database/rpm_metadata.go b/internal/database/rpm_metadata.go index 2466b06..ccde123 100644 --- a/internal/database/rpm_metadata.go +++ b/internal/database/rpm_metadata.go @@ -40,6 +40,13 @@ func (db *DB) DeleteRPMMetadata(ctx context.Context, repoName, filePath string) return err } +// RPMMetadataFingerprint is a cheap change detector for a repo's rows: they are +// only ever inserted or deleted, so (count, max id) moves whenever they change. +func (db *DB) RPMMetadataFingerprint(ctx context.Context, repoName string) (count, maxID int64, err error) { + err = db.Pool.QueryRow(ctx, `SELECT count(*), coalesce(max(id), 0) FROM rpm_metadata WHERE repo_name = $1`, repoName).Scan(&count, &maxID) + return count, maxID, err +} + type RPMMetadataRow struct { RepoName string FilePath string diff --git a/internal/provider/rpm/github.go b/internal/provider/rpm/github.go index 707b54d..e459690 100644 --- a/internal/provider/rpm/github.go +++ b/internal/provider/rpm/github.go @@ -15,9 +15,11 @@ import ( "regexp" "strings" "sync" + "sync/atomic" "time" rpmlib "github.com/cavaliergopher/rpm" + "golang.org/x/sync/singleflight" "golang.org/x/time/rate" "git.unkin.net/unkin/artifactapi/internal/githubauth" @@ -52,6 +54,9 @@ const ( // retryable 503. Kept short so a client never hangs on a rate-limited derive // 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 + repodataSweepEvery = time.Minute ) // GitHubProvider is a metadata-only remote: it scans a GitHub repo's releases @@ -84,6 +89,26 @@ type GitHubProvider struct { mu sync.Mutex scanning map[string]bool lastScan map[string]time.Time + + // repodata caches generated indexes for cache_repodata remotes, keyed by remote name. + 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. +// Rows are insert-or-delete only, so (count, max id) changes whenever they do. +type rpmFingerprinter interface { + RPMMetadataFingerprint(ctx context.Context, repoName string) (count, maxID int64, err error) +} + +type cachedRepodata struct { + count, maxID int64 + seq int64 // build order, so a slow older build never replaces a newer one + used time.Time + *repodata } func newGitHubProvider() *GitHubProvider { @@ -97,6 +122,7 @@ func newGitHubProvider() *GitHubProvider { coldWait: defaultColdWait, scanning: map[string]bool{}, lastScan: map[string]time.Time{}, + repodataCache: map[string]cachedRepodata{}, } } @@ -170,6 +196,11 @@ func (p *GitHubProvider) ServeRemote(w http.ResponseWriter, r *http.Request, rem } tail := strings.TrimPrefix(path, "repodata/") + if fp, ok := store.(rpmFingerprinter); ok && remote.CacheRepodata { + p.serveCachedRepodata(w, sr, store, fp, remote.Name, tail) + return true + } + p.dropRepodata(remote.Name) lp := &Provider{} switch { case tail == "repomd.xml": @@ -239,7 +270,104 @@ func (p *GitHubProvider) ensurePrimed(ctx context.Context, remote models.Remote, return false } +// serveCachedRepodata serves a file from the generated index, rebuilding it +// only when the rows' fingerprint moved. The fingerprint is read from the shared +// 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) { + var pick func(*repodata) []byte + contentType := "application/gzip" + switch { + case tail == "repomd.xml": + pick, contentType = func(rd *repodata) []byte { return rd.repomd }, "application/xml" + case strings.HasSuffix(tail, "-primary.xml.gz"): + pick = func(rd *repodata) []byte { return rd.primary } + case strings.HasSuffix(tail, "-filelists.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(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 + } + now := time.Now() + p.repodataMu.Lock() + p.sweepRepodataLocked(now) + c, ok := p.repodataCache[name] + if ok { + c.used = now + p.repodataCache[name] = c + } + p.repodataMu.Unlock() + if ok && c.count == count && c.maxID == maxID { + return c.repodata, nil + } + + v, err, _ := p.repodataGroup.Do(fmt.Sprintf("%s\x00%d\x00%d", name, count, maxID), func() (any, error) { + metas, err := store.ListRPMMetadataEntries(ctx, name) + if err != nil { + return nil, err + } + // Rows read after the fingerprint can only be newer than it, so the + // worst case is one extra rebuild on the next request. + rd := buildRepodata(metas) + p.repodataMu.Lock() + 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 + }) + if err != nil { + return nil, err + } + return v.(*repodata), nil +} + +func (p *GitHubProvider) dropRepodata(name string) { + p.repodataMu.Lock() + delete(p.repodataCache, name) + p.sweepRepodataLocked(time.Now()) + 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) + } + } +} + func (p *GitHubProvider) cacheEmpty(ctx context.Context, store provider.RemoteMetadataStore, name string) bool { + if fp, ok := store.(rpmFingerprinter); ok { + count, _, err := fp.RPMMetadataFingerprint(ctx, name) + return err == nil && count == 0 + } rows, err := store.ListRPMMetadataEntries(ctx, name) if err != nil { // Treat a failed read as "not empty" so a transient DB error becomes a diff --git a/internal/provider/rpm/github_repodata_cache_test.go b/internal/provider/rpm/github_repodata_cache_test.go new file mode 100644 index 0000000..6cd5388 --- /dev/null +++ b/internal/provider/rpm/github_repodata_cache_test.go @@ -0,0 +1,347 @@ +package rpm + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "git.unkin.net/unkin/artifactapi/internal/provider" + "git.unkin.net/unkin/artifactapi/pkg/models" +) + +// fpStore adds a fingerprint and a full-read counter to fakeStore. +type fpStore struct { + *fakeStore + 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) +} + +func (s *fpStore) InsertRPMMetadata(ctx context.Context, m *provider.RPMMetadata) error { + s.mu.Lock() + _, exists := s.rows[m.FilePath] + s.mu.Unlock() + if err := s.fakeStore.InsertRPMMetadata(ctx, m); err != nil || exists { + return err // ON CONFLICT DO NOTHING leaves max(id) unchanged + } + s.mu.Lock() + s.seq++ + s.mu.Unlock() + return nil +} + +func (s *fpStore) RPMMetadataFingerprint(_ context.Context, _ string) (int64, int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + 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 { + s.mu.Lock() + defer s.mu.Unlock() + return s.lists +} + +func newCacheFixture(t *testing.T, cache bool) (*GitHubProvider, *fpStore, models.Remote) { + t.Helper() + remote := models.Remote{Name: "cached-rpm", PackageType: models.PackageGitHubRPM, MutableTTL: 3600, CacheRepodata: cache} + p := newGitHubProvider() + p.lastScan[remote.Name] = time.Now() + store := &fpStore{fakeStore: newFakeStore()} + addPkg(t, store, "alpha") + return p, store, remote +} + +func addPkg(t *testing.T, store *fpStore, name string) { + t.Helper() + if err := store.InsertRPMMetadata(context.Background(), &provider.RPMMetadata{ + RepoName: "cached-rpm", FilePath: name + "-1-1.noarch.rpm", ContentHash: "sha256:" + name, + Name: name, Version: "1", Release: "1", Arch: "noarch", Group: "Unspecified", + Files: []provider.RPMFile{{Path: "/usr/bin/" + name}}, + }); err != nil { + t.Fatal(err) + } +} + +func serveReq(t *testing.T, p *GitHubProvider, remote models.Remote, store *fpStore, req *http.Request, path string) *httptest.ResponseRecorder { + t.Helper() + rec := httptest.NewRecorder() + if !p.ServeRemote(rec, req, remote, path, "https://proxy.example", store) { + 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()) + } + return rec.Body.Bytes() +} + +var repodataPaths = []string{ + "repodata/repomd.xml", + "repodata/x-primary.xml.gz", + "repodata/x-filelists.xml.gz", + "repodata/x-other.xml.gz", +} + +func TestCachedRepodataHitSkipsRegeneration(t *testing.T) { + p, store, remote := newCacheFixture(t, true) + + repomd := serve(t, p, remote, store, "repodata/repomd.xml") + primary := serve(t, p, remote, store, "repodata/x-primary.xml.gz") + if n := store.listCount(); n != 1 { + t.Fatalf("full metadata reads = %d, want 1", n) + } + if !strings.Contains(string(repomd), sha256Hex(primary)+"-primary.xml.gz") { + t.Fatalf("cached repomd does not reference cached primary:\n%s", repomd) + } +} + +func TestCachedRepodataRebuildsOnDataChange(t *testing.T) { + p, store, remote := newCacheFixture(t, true) + before := serve(t, p, remote, store, "repodata/repomd.xml") + + addPkg(t, store, "beta") + after := serve(t, p, remote, store, "repodata/repomd.xml") + primary := serve(t, p, remote, store, "repodata/x-primary.xml.gz") + + if string(before) == string(after) { + t.Fatal("repomd unchanged after metadata change") + } + if !strings.Contains(string(after), sha256Hex(primary)+"-primary.xml.gz") { + t.Fatal("rebuilt repomd does not reference rebuilt primary") + } + if !strings.Contains(gunzip(t, primary), "beta") { + t.Fatal("rebuilt primary missing new package") + } + if n := store.listCount(); n != 2 { + t.Fatalf("full metadata reads = %d, want 2", n) + } +} + +func TestUncachedRepodataMatchesCachedAndReadsEveryTime(t *testing.T) { + p, store, remote := newCacheFixture(t, true) + cached := map[string]string{} + for _, path := range repodataPaths { + cached[path] = string(serve(t, p, remote, store, path)) + } + + remote.CacheRepodata = false + reads := store.listCount() + for _, path := range repodataPaths { + if got := string(serve(t, p, remote, store, path)); got != cached[path] { + t.Fatalf("%s differs between cached and uncached", path) + } + } + if n := store.listCount() - reads; n != len(repodataPaths) { + t.Fatalf("uncached full reads = %d, want %d", n, len(repodataPaths)) + } + if _, ok := p.repodataCache[remote.Name]; ok { + t.Fatal("disabling cache_repodata should drop the cached index") + } +} + +func TestCachedRepodataRebuildsOnSameCountReinsert(t *testing.T) { + p, store, remote := newCacheFixture(t, true) + addPkg(t, store, "alpha") // conflicting insert: fingerprint must not move + serve(t, p, remote, store, "repodata/repomd.xml") + serve(t, p, remote, store, "repodata/repomd.xml") + if n := store.listCount(); n != 1 { + t.Fatalf("full reads after conflicting insert = %d, want 1", n) + } + + if err := store.DeleteRPMMetadata(context.Background(), remote.Name, "alpha-1-1.noarch.rpm"); err != nil { + t.Fatal(err) + } + addPkg(t, store, "alpha") + serve(t, p, remote, store, "repodata/repomd.xml") + if n := store.listCount(); n != 2 { + t.Fatalf("full reads after same-count reinsert = %d, want 2", n) + } +} + +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 { + t.Fatal("idle remote's index not evicted") + } + if _, ok := p.repodataCache[remote.Name]; !ok { + 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 a0f279b..5e18fcb 100644 --- a/internal/provider/rpm/rpm.go +++ b/internal/provider/rpm/rpm.go @@ -250,36 +250,52 @@ 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 { return } - primary := generatePrimaryXMLGZ(metas) - filelists := generateFilelistsXMLGZ(metas) - other := generateOtherXMLGZ(metas) - - primaryHash := sha256Hex(primary) - filelistsHash := sha256Hex(filelists) - otherHash := sha256Hex(other) - - repomd := generateRepomd(repomdRevision(metas), primaryHash, len(primary), filelistsHash, len(filelists), otherHash, len(other)) - w.Header().Set("Content-Type", "application/xml") w.WriteHeader(http.StatusOK) - w.Write(repomd) + _, _ = w.Write(buildRepodata(metas).repomd) +} + +// repodata is a complete generated index: repomd.xml plus the three gz files it references. +type repodata struct { + repomd, primary, filelists, other []byte +} + +func buildRepodata(metas []provider.RPMMetadata) *repodata { + rd := &repodata{ + primary: generatePrimaryXMLGZ(metas), + filelists: generateFilelistsXMLGZ(metas), + other: generateOtherXMLGZ(metas), + } + rd.repomd = generateRepomd(repomdRevision(metas), + sha256Hex(rd.primary), len(rd.primary), + sha256Hex(rd.filelists), len(rd.filelists), + sha256Hex(rd.other), len(rd.other)) + return rd } func (p *Provider) servePrimary(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) { diff --git a/internal/tui/render.go b/internal/tui/render.go index 635a8ca..c18b260 100644 --- a/internal/tui/render.go +++ b/internal/tui/render.go @@ -59,6 +59,7 @@ func (m *model) viewRemoteDetail() string { sb.WriteString(fmt.Sprintf(" Revalidation: %v\n", r.CheckMutable)) sb.WriteString(fmt.Sprintf(" Stale on Error: %v\n", r.StaleOnError)) fmt.Fprintf(&sb, " Serve Stale: %v\n", r.ServeStale) + fmt.Fprintf(&sb, " Cache Repodata: %v\n", r.CacheRepodata) if len(r.Patterns) > 0 { sb.WriteString(fmt.Sprintf(" Patterns: %s\n", strings.Join(r.Patterns, ", "))) diff --git a/migrations/0010_remote_cache_repodata.sql b/migrations/0010_remote_cache_repodata.sql new file mode 100644 index 0000000..8fb7297 --- /dev/null +++ b/migrations/0010_remote_cache_repodata.sql @@ -0,0 +1 @@ +ALTER TABLE remotes ADD COLUMN IF NOT EXISTS cache_repodata BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/pkg/models/remote.go b/pkg/models/remote.go index a230fd0..6a2c3b8 100644 --- a/pkg/models/remote.go +++ b/pkg/models/remote.go @@ -84,6 +84,10 @@ type Remote struct { ReleasesRemote string `json:"releases_remote,omitempty"` ManagedBy string `json:"managed_by,omitempty"` + // CacheRepodata keeps a github_rpm remote's generated repodata in memory, + // rebuilding it only when its metadata rows change. + CacheRepodata bool `json:"cache_repodata"` + CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } diff --git a/ui/src/api/types.ts b/ui/src/api/types.ts index 5bc4257..bc6b020 100644 --- a/ui/src/api/types.ts +++ b/ui/src/api/types.ts @@ -19,6 +19,7 @@ export interface Remote { quarantine_days: number; stale_on_error: boolean; serve_stale: boolean; + cache_repodata: boolean; releases_remote: string; managed_by: string; created_at: string; diff --git a/ui/src/pages/RemoteDetail.tsx b/ui/src/pages/RemoteDetail.tsx index 96cb03b..875a6f0 100644 --- a/ui/src/pages/RemoteDetail.tsx +++ b/ui/src/pages/RemoteDetail.tsx @@ -52,6 +52,8 @@ export function RemoteDetail() {
{remote.stale_on_error ? 'enabled' : 'disabled'}
Serve Stale
{remote.serve_stale ? 'enabled' : 'disabled'}
+
Cache Repodata
+
{remote.cache_repodata ? 'enabled' : 'disabled'}
{remote.releases_remote && ( <>
Releases Remote