unkin/artifactapi
3
0
Fork 0
Code Issues 5 Pull Requests 9 Actions Packages Projects Releases 1 Wiki Activity

Share the virtual merge cache across index types (#147)

Browse Source 
The rpm virtual's merge cache, singleflight and two-generation retention are hard-wired to RPMRepo, so other merged index types (deb next) can't reuse them.

- add a mergedRepo interface (index + referenced files) implemented by RPMRepo
- key the merge cache by type-scoped string and serve files from both retained generations via cachedFile
- extract parallel fail-closed member fetching into a generic fetchMembers
- add tests for generation retention and per-key isolation

Reviewed-on: #147
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
This commit was merged in pull request #147.
This commit is contained in:
Unkin Agent
2026-10-17 13:28:15 +11:00
committed by BenVincent
parent 9b91aa39a6
commit 0150b5ebb4
2 changed files with 105 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/virtual/engine.go
+55 -35
View File
@@ -29,15 +29,24 @@ type Engine struct {
mergeTTL time.Duration
sf singleflight.Group
mu sync.Mutex
rpm map[string]*rpmGen
sf singleflight.Group
mu sync.Mutex
gens map[string]*repoGen
}
// rpmGen holds a virtual's current merge plus the previous one, so a client
// holding the previous repomd can still fetch its content-hashed data files.
type rpmGen struct {
cur, prev *RPMRepo
// mergedRepo is a merged index document plus the files it references.
type mergedRepo interface {
index() []byte
files() map[string][]byte
}
func (r *RPMRepo) index() []byte { return r.Repomd }
func (r *RPMRepo) files() map[string][]byte { return r.Files }
// repoGen holds a virtual's current merge plus the previous one, so a client
// holding the previous index can still fetch the files it references.
type repoGen struct {
cur, prev mergedRepo
at time.Time
}
@@ -217,54 +226,57 @@ func (e *Engine) fetchRPM(ctx context.Context, virt models.Virtual, path string)
if !strings.HasPrefix(path, "repodata/") {
return nil, "", ErrNotFound
}
key := "rpm\x00" + virt.Name
if path != "repodata/repomd.xml" {
if body, ok := e.cachedRPMFile(virt.Name, path); ok {
if body, ok := e.cachedFile(key, path); ok {
return body, "application/gzip", nil
}
}
repo, err := e.mergedRPM(ctx, virt)
repo, err := e.merged(ctx, key, func(ctx context.Context) (mergedRepo, error) {
return e.mergeRPM(ctx, virt)
})
if err != nil {
return nil, "", err
}
if path == "repodata/repomd.xml" {
return repo.Repomd, "application/xml", nil
return repo.index(), "application/xml", nil
}
if body, ok := repo.Files[path]; ok {
if body, ok := repo.files()[path]; ok {
return body, "application/gzip", nil
}
return nil, "", ErrNotFound
}
// mergedRPM returns the virtual's merge, reusing it for mergeTTL and
// collapsing concurrent merges of the same virtual into one.
// merged returns the merge cached under key, reusing it for mergeTTL and
// collapsing concurrent merges of the same key into one.
// ponytail: per-replica cache; behind a non-sticky LB a data request landing on
// another replica re-merges and 404s if a member changed in between. Move to the
// shared redis cache if that shows up.
func (e *Engine) mergedRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
func (e *Engine) merged(ctx context.Context, key string, merge func(context.Context) (mergedRepo, error)) (mergedRepo, error) {
e.mu.Lock()
g := e.rpm[virt.Name]
g := e.gens[key]
if g != nil && time.Since(g.at) < e.mergeTTL {
e.mu.Unlock()
return g.cur, nil
}
e.mu.Unlock()
v, err, _ := e.sf.Do(virt.Name, func() (any, error) {
repo, err := e.mergeRPM(context.WithoutCancel(ctx), virt)
v, err, _ := e.sf.Do(key, func() (any, error) {
repo, err := merge(context.WithoutCancel(ctx))
if err != nil {
return nil, err
}
e.mu.Lock()
defer e.mu.Unlock()
if e.rpm == nil {
e.rpm = map[string]*rpmGen{}
if e.gens == nil {
e.gens = map[string]*repoGen{}
}
g := e.rpm[virt.Name]
g := e.gens[key]
switch {
case g == nil:
e.rpm[virt.Name] = &rpmGen{cur: repo, at: time.Now()}
case string(g.cur.Repomd) == string(repo.Repomd):
e.gens[key] = &repoGen{cur: repo, at: time.Now()}
case string(g.cur.index()) == string(repo.index()):
g.at = time.Now()
repo = g.cur
default:
@@ -275,18 +287,31 @@ func (e *Engine) mergedRPM(ctx context.Context, virt models.Virtual) (*RPMRepo,
if err != nil {
return nil, err
}
return v.(*RPMRepo), nil
return v.(mergedRepo), nil
}
func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
members := make([]RPMMember, len(virt.Members))
members, err := fetchMembers(virt, func(name string) (*RPMMember, error) { return e.rpmMember(ctx, name) })
if err != nil {
return nil, err
}
repo, err := MergeRPM(members)
if err != nil {
return nil, fmt.Errorf("merge rpm repodata: %w", err)
}
return repo, nil
}
// fetchMembers reads every member in parallel, failing if any member fails.
func fetchMembers[T any](virt models.Virtual, fetch func(name string) (*T, error)) ([]T, error) {
members := make([]T, len(virt.Members))
errs := make([]error, len(virt.Members))
var wg sync.WaitGroup
for i, name := range virt.Members {
wg.Add(1)
go func() {
defer wg.Done()
m, err := e.rpmMember(ctx, name)
m, err := fetch(name)
if err != nil {
errs[i] = fmt.Errorf("member %q: %w", name, err)
return
@@ -299,26 +324,21 @@ func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, e
if err := errors.Join(errs...); err != nil {
return nil, fmt.Errorf("virtual %q: %v", virt.Name, err)
}
repo, err := MergeRPM(members)
if err != nil {
return nil, fmt.Errorf("merge rpm repodata: %w", err)
}
return repo, nil
return members, nil
}
func (e *Engine) cachedRPMFile(virt, path string) ([]byte, bool) {
func (e *Engine) cachedFile(key, path string) ([]byte, bool) {
e.mu.Lock()
defer e.mu.Unlock()
g := e.rpm[virt]
g := e.gens[key]
if g == nil {
return nil, false
}
for _, r := range []*RPMRepo{g.cur, g.prev} {
for _, r := range []mergedRepo{g.cur, g.prev} {
if r == nil {
continue
}
if body, ok := r.Files[path]; ok {
if body, ok := r.files()[path]; ok {
return body, true
}
}
internal/virtual/engine_test.go
+50
View File
@@ -156,3 +156,53 @@ func TestMemberRedirect(t *testing.T) {
}
}
}
type fakeRepo struct {
idx string
fs map[string][]byte
}
func (r *fakeRepo) index() []byte { return []byte(r.idx) }
func (r *fakeRepo) files() map[string][]byte { return r.fs }
func genRepo(gen string) func(context.Context) (mergedRepo, error) {
return func(context.Context) (mergedRepo, error) {
return &fakeRepo{idx: gen, fs: map[string][]byte{gen + ".gz": []byte(gen), "fixed": []byte(gen)}}, nil
}
}
func TestMergedRetainsPreviousGeneration(t *testing.T) {
e := fakeEngine(nil)
for _, gen := range []string{"g1", "g2", "g3"} {
if _, err := e.merged(context.Background(), "k", genRepo(gen)); err != nil {
t.Fatal(err)
}
}
for path, want := range map[string]string{"g3.gz": "g3", "g2.gz": "g2", "fixed": "g3"} {
if body, ok := e.cachedFile("k", path); !ok || string(body) != want {
t.Errorf("%s: got (%q, %v), want %q", path, body, ok, want)
}
}
if _, ok := e.cachedFile("k", "g1.gz"); ok {
t.Error("generation two merges back must be dropped")
}
}
func TestMergedReusesWithinTTLAndIsolatesKeys(t *testing.T) {
e := fakeEngine(nil)
e.mergeTTL = time.Minute
for _, step := range []struct{ key, gen, want string }{
{"a", "a1", "a1"}, {"b", "b1", "b1"}, {"a", "a2", "a1"},
} {
repo, err := e.merged(context.Background(), step.key, genRepo(step.gen))
if err != nil || string(repo.index()) != step.want {
t.Fatalf("%s: got %v (err %v), want %s", step.key, repo, err, step.want)
}
}
if _, ok := e.cachedFile("b", "a1.gz"); ok {
t.Error("key b served key a's file")
}
if _, ok := e.cachedFile("c", "a1.gz"); ok {
t.Error("unknown key served a file")
}
}