Add parallel_fetch toggle for virtual member index files (#165)

Virtual members already merge concurrently, but each rpm member fetches primary/filelists/other one after another, and each deb member its Packages files the same way. On large virtuals the slowest member's serial chain pushes the merge past dnf's 30s timeout.

- add `parallel_fetch` (default false) to virtuals: model, v2 API, migration 0007
- fetch a member's index files concurrently (up to 4) when set; first error still fails the member
- thread the flag through rpm and deb member fetches

Reviewed-on: #165
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>

internal/api/v2/virtuals_test.go

@@ -0,0 +1,56 @@
+package v2
+
+import (
+	"encoding/json"
+	"net/http/httptest"
+	"testing"
+
+	"git.unkin.net/unkin/artifactapi/internal/database"
+)
+
+func TestVirtualParallelFetchRoundTrip(t *testing.T) {
+	if testDSN == "" {
+		t.Skip("Docker unavailable")
+	}
+	db, err := database.New(testDSN)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	h := NewVirtualsHandler(db).Routes()
+	parallel := func() bool {
+		t.Helper()
+		w := httptest.NewRecorder()
+		h.ServeHTTP(w, httptest.NewRequest("GET", "/vpf", nil))
+		var got map[string]any
+		if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || w.Code != 200 {
+			t.Fatalf("get = %d %s", w.Code, w.Body)
+		}
+		v, ok := got["parallel_fetch"].(bool)
+		if !ok {
+			t.Fatalf("parallel_fetch missing from %s", w.Body)
+		}
+		return v
+	}
+	const body = `"package_type":"rpm","members":["a"]`
+
+	steps := []struct {
+		method, path, json string
+		want               bool
+	}{
+		{"POST", "/", `{"name":"vpf",` + body + `}`, false},
+		{"PUT", "/vpf", `{` + body + `,"parallel_fetch":true}`, true},
+		{"PUT", "/vpf", `{` + body + `,"parallel_fetch":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 := parallel(); got != s.want {
+			t.Errorf("step %d: parallel_fetch = %v, want %v", i, got, s.want)
+		}
+	}
+	if c := do(t, h, "DELETE", "/vpf", ""); c != 204 {
+		t.Fatalf("delete = %d", c)
+	}
+}

internal/database/database_test.go

@@ -335,6 +335,40 @@ func TestDeleteRemoteDropsLocalFiles(t *testing.T) {
 	}
 }
 
+func TestVirtualParallelFetchRoundTrip(t *testing.T) {
+	requireDB(t)
+	if err := testDB.CreateVirtual(ctx(), &models.Virtual{
+		Name: "v-parallel", PackageType: models.PackageRPM, Members: []string{"a"}, ParallelFetch: true,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = testDB.DeleteVirtual(ctx(), "v-parallel") }()
+	v, err := testDB.GetVirtual(ctx(), "v-parallel")
+	if err != nil || !v.ParallelFetch {
+		t.Fatalf("created parallel_fetch = %v, %v; want true", v, err)
+	}
+	v.ParallelFetch = false
+	if err := testDB.UpdateVirtual(ctx(), v); err != nil {
+		t.Fatal(err)
+	}
+	vs, err := testDB.ListVirtuals(ctx())
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, l := range vs {
+		if l.Name == "v-parallel" && l.ParallelFetch {
+			t.Fatal("updated parallel_fetch not persisted")
+		}
+	}
+	if err := testDB.CreateVirtual(ctx(), &models.Virtual{Name: "v-serial", PackageType: models.PackageRPM, Members: []string{"a"}}); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = testDB.DeleteVirtual(ctx(), "v-serial") }()
+	if v, err := testDB.GetVirtual(ctx(), "v-serial"); err != nil || v.ParallelFetch {
+		t.Fatalf("default parallel_fetch = %v, %v; want false", v, err)
+	}
+}
+
 func TestVirtualsCRUD(t *testing.T) {
 	requireDB(t)
 	if err := testDB.CreateVirtual(ctx(), &models.Virtual{

internal/database/virtuals.go

@@ -8,12 +8,12 @@ import (
 
 func (db *DB) GetVirtual(ctx context.Context, name string) (*models.Virtual, error) {
 	row := db.Pool.QueryRow(ctx, `
-		SELECT name, package_type, description, members, managed_by, created_at, updated_at
+		SELECT name, package_type, description, members, managed_by, parallel_fetch, created_at, updated_at
 		FROM virtuals WHERE name = $1
 	`, name)
 
 	var v models.Virtual
-	err := row.Scan(&v.Name, &v.PackageType, &v.Description, &v.Members, &v.ManagedBy, &v.CreatedAt, &v.UpdatedAt)
+	err := row.Scan(&v.Name, &v.PackageType, &v.Description, &v.Members, &v.ManagedBy, &v.ParallelFetch, &v.CreatedAt, &v.UpdatedAt)
 	if err != nil {
 		return nil, err
 	}
@@ -22,7 +22,7 @@ func (db *DB) GetVirtual(ctx context.Context, name string) (*models.Virtual, err
 
 func (db *DB) ListVirtuals(ctx context.Context) ([]models.Virtual, error) {
 	rows, err := db.Pool.Query(ctx, `
-		SELECT name, package_type, description, members, managed_by, created_at, updated_at
+		SELECT name, package_type, description, members, managed_by, parallel_fetch, created_at, updated_at
 		FROM virtuals ORDER BY name
 	`)
 	if err != nil {
@@ -33,7 +33,7 @@ func (db *DB) ListVirtuals(ctx context.Context) ([]models.Virtual, error) {
 	var virtuals []models.Virtual
 	for rows.Next() {
 		var v models.Virtual
-		if err := rows.Scan(&v.Name, &v.PackageType, &v.Description, &v.Members, &v.ManagedBy, &v.CreatedAt, &v.UpdatedAt); err != nil {
+		if err := rows.Scan(&v.Name, &v.PackageType, &v.Description, &v.Members, &v.ManagedBy, &v.ParallelFetch, &v.CreatedAt, &v.UpdatedAt); err != nil {
 			return nil, err
 		}
 		virtuals = append(virtuals, v)
@@ -43,18 +43,18 @@ func (db *DB) ListVirtuals(ctx context.Context) ([]models.Virtual, error) {
 
 func (db *DB) CreateVirtual(ctx context.Context, v *models.Virtual) error {
 	_, err := db.Pool.Exec(ctx, `
-		INSERT INTO virtuals (name, package_type, description, members, managed_by)
-		VALUES ($1, $2, $3, $4, $5)
-	`, v.Name, v.PackageType, v.Description, v.Members, v.ManagedBy)
+		INSERT INTO virtuals (name, package_type, description, members, managed_by, parallel_fetch)
+		VALUES ($1, $2, $3, $4, $5, $6)
+	`, v.Name, v.PackageType, v.Description, v.Members, v.ManagedBy, v.ParallelFetch)
 	return err
 }
 
 func (db *DB) UpdateVirtual(ctx context.Context, v *models.Virtual) error {
 	_, err := db.Pool.Exec(ctx, `
 		UPDATE virtuals SET
-			package_type=$2, description=$3, members=$4, managed_by=$5, updated_at=NOW()
+			package_type=$2, description=$3, members=$4, managed_by=$5, parallel_fetch=$6, updated_at=NOW()
 		WHERE name=$1
-	`, v.Name, v.PackageType, v.Description, v.Members, v.ManagedBy)
+	`, v.Name, v.PackageType, v.Description, v.Members, v.ManagedBy, v.ParallelFetch)
 	return err
 }
 

internal/virtual/engine.go

@@ -19,6 +19,7 @@ import (
 	"git.unkin.net/unkin/artifactapi/internal/provider"
 	"git.unkin.net/unkin/artifactapi/internal/proxy"
 	"git.unkin.net/unkin/artifactapi/pkg/models"
+	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/singleflight"
 )
 
@@ -26,10 +27,11 @@ type Engine struct {
 	db          *database.DB
 	proxyEngine *proxy.Engine
 	getRemote   func(context.Context, string) (*models.Remote, error)
-	rpmMember   func(context.Context, string) (*RPMMember, error)
-	debMember   func(ctx context.Context, name, base string) (*DebMember, error)
+	rpmMember   func(ctx context.Context, name string, parallel bool) (*RPMMember, error)
+	debMember   func(ctx context.Context, name, base string, parallel bool) (*DebMember, error)
 	apkMember   func(ctx context.Context, name, dir string) (*AlpineMember, error)
 	npmMember   func(ctx context.Context, name, pkg string) (*NPMMember, error)
+	memberPath  func(ctx context.Context, remote models.Remote, path string) ([]byte, error)
 
 	// NPMLocal returns a local npm repo's packument with tarball paths relative
 	// to the repo, or ErrNotFound.
@@ -91,6 +93,7 @@ func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine {
 	e.ansibleDoc = e.ansibleMemberDoc
 	e.goMember = e.fetchGoMember
 	e.puppetDoc = e.fetchMemberPath
+	e.memberPath = e.fetchMemberPath
 	return e
 }
 
@@ -353,7 +356,7 @@ func (e *Engine) merged(ctx context.Context, key string, merge func(context.Cont
 }
 
 func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
-	members, err := fetchMembers(virt, func(name string) (*RPMMember, error) { return e.rpmMember(ctx, name) })
+	members, err := fetchMembers(virt, func(name string) (*RPMMember, error) { return e.rpmMember(ctx, name, virt.ParallelFetch) })
 	if err != nil {
 		return nil, err
 	}
@@ -402,7 +405,7 @@ func (e *Engine) fetchDeb(ctx context.Context, virt models.Virtual, reqPath stri
 	}
 
 	repo, err := e.merged(ctx, key, func(ctx context.Context) (mergedRepo, error) {
-		members, err := fetchMembers(virt, func(name string) (*DebMember, error) { return e.debMember(ctx, name, base) })
+		members, err := fetchMembers(virt, func(name string) (*DebMember, error) { return e.debMember(ctx, name, base, virt.ParallelFetch) })
 		if err != nil {
 			return nil, err
 		}
@@ -430,26 +433,22 @@ func debContentType(name string) string {
 
 // fetchDebMember reads a member's Release under base and each binary Packages
 // index it lists.
-func (e *Engine) fetchDebMember(ctx context.Context, name, base string) (*DebMember, error) {
+func (e *Engine) fetchDebMember(ctx context.Context, name, base string, parallel bool) (*DebMember, error) {
 	remote, err := e.getRemote(ctx, name)
 	if err != nil {
 		return nil, fmt.Errorf("remote %q: %w", name, err)
 	}
-	rel, err := e.fetchMemberPath(ctx, *remote, base+"Release")
+	rel, err := e.memberPath(ctx, *remote, base+"Release")
 	if err != nil {
 		return nil, err
 	}
-	m := &DebMember{RemoteName: name, Release: rel, Packages: map[string][]byte{}}
-	for dir, file := range debIndexFiles(parseDebRelease(rel)) {
-		raw, err := e.fetchMemberPath(ctx, *remote, base+file)
-		if err != nil {
-			return nil, err
-		}
-		if m.Packages[dir], err = decompress(file, raw); err != nil {
-			return nil, fmt.Errorf("decompress %s: %w", file, err)
-		}
+	pkgs, err := fetchFiles(ctx, parallel, debIndexFiles(parseDebRelease(rel)), func(ctx context.Context, file string) ([]byte, error) {
+		return e.fetchDecompressed(ctx, *remote, base, file)
+	})
+	if err != nil {
+		return nil, err
 	}
-	return m, nil
+	return &DebMember{RemoteName: name, Release: rel, Packages: pkgs}, nil
 }
 
 func (e *Engine) mergedAlpine(ctx context.Context, virt models.Virtual, dir string) (mergedRepo, error) {
@@ -559,12 +558,12 @@ func (e *Engine) cachedFile(key, path string) ([]byte, bool) {
 	return nil, false
 }
 
-func (e *Engine) fetchRPMMember(ctx context.Context, name string) (*RPMMember, error) {
+func (e *Engine) fetchRPMMember(ctx context.Context, name string, parallel bool) (*RPMMember, error) {
 	remote, err := e.getRemote(ctx, name)
 	if err != nil {
 		return nil, fmt.Errorf("remote %q: %w", name, err)
 	}
-	repomd, err := e.fetchMemberPath(ctx, *remote, "repodata/repomd.xml")
+	repomd, err := e.memberPath(ctx, *remote, "repodata/repomd.xml")
 	if err != nil {
 		return nil, err
 	}
@@ -572,17 +571,66 @@ func (e *Engine) fetchRPMMember(ctx context.Context, name string) (*RPMMember, e
 	if err != nil {
 		return nil, err
 	}
-	m := &RPMMember{RemoteName: name, Bases: remote.UpstreamPool(), Timestamp: ts, Data: map[string][]byte{}}
-	for t, href := range locs {
-		raw, err := e.fetchMemberPath(ctx, *remote, href)
-		if err != nil {
-			return nil, err
-		}
-		if m.Data[t], err = decompress(href, raw); err != nil {
-			return nil, fmt.Errorf("decompress %s: %w", href, err)
-		}
+	data, err := fetchFiles(ctx, parallel, locs, func(ctx context.Context, href string) ([]byte, error) {
+		return e.fetchDecompressed(ctx, *remote, "", href)
+	})
+	if err != nil {
+		return nil, err
 	}
-	return m, nil
+	return &RPMMember{RemoteName: name, Bases: remote.UpstreamPool(), Timestamp: ts, Data: data}, nil
+}
+
+func (e *Engine) fetchDecompressed(ctx context.Context, remote models.Remote, base, file string) ([]byte, error) {
+	raw, err := e.memberPath(ctx, remote, base+file)
+	if err != nil {
+		return nil, err
+	}
+	body, err := decompress(file, raw)
+	if err != nil {
+		return nil, fmt.Errorf("decompress %s: %w", file, err)
+	}
+	return body, nil
+}
+
+// maxFileFetches bounds the concurrent index file fetches per member.
+const maxFileFetches = 4
+
+// fetchFiles fetches each key's file, one at a time or up to maxFileFetches at
+// once, failing on the first error and cancelling the rest.
+func fetchFiles(ctx context.Context, parallel bool, files map[string]string, fetch func(ctx context.Context, file string) ([]byte, error)) (map[string][]byte, error) {
+	out := make(map[string][]byte, len(files))
+	if !parallel {
+		for k, f := range files {
+			body, err := fetch(ctx, f)
+			if err != nil {
+				return nil, err
+			}
+			out[k] = body
+		}
+		return out, nil
+	}
+	var mu sync.Mutex
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(maxFileFetches)
+	for k, f := range files {
+		g.Go(func() error {
+			if err := gctx.Err(); err != nil {
+				return err
+			}
+			body, err := fetch(gctx, f)
+			if err != nil {
+				return err
+			}
+			mu.Lock()
+			out[k] = body
+			mu.Unlock()
+			return nil
+		})
+	}
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+	return out, nil
 }
 
 // fetchMemberPath reads one path from a member the same way its own route

internal/virtual/engine_test.go

@@ -19,7 +19,7 @@ func fakeEngine(members map[string]*RPMMember) *Engine {
 		getRemote: func(_ context.Context, name string) (*models.Remote, error) {
 			return &models.Remote{Name: name, RepoType: models.RepoTypeRemote}, nil
 		},
-		rpmMember: func(_ context.Context, name string) (*RPMMember, error) {
+		rpmMember: func(_ context.Context, name string, _ bool) (*RPMMember, error) {
 			if m := members[name]; m != nil {
 				return m, nil
 			}
@@ -75,7 +75,7 @@ func TestFetchRPMLocalMemberWithoutRepodataIsUpstreamError(t *testing.T) {
 	e := fakeEngine(map[string]*RPMMember{
 		"a": {RemoteName: "a", Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}},
 	})
-	e.rpmMember = func(_ context.Context, name string) (*RPMMember, error) {
+	e.rpmMember = func(_ context.Context, name string, _ bool) (*RPMMember, error) {
 		if name == "local" {
 			return nil, fmt.Errorf("local/repodata/repomd.xml: %w", ErrNotFound)
 		}
@@ -91,7 +91,7 @@ func TestFetchRPMMergeIsShared(t *testing.T) {
 	var calls atomic.Int32
 	e := fakeEngine(nil)
 	e.mergeTTL = time.Minute
-	e.rpmMember = func(_ context.Context, name string) (*RPMMember, error) {
+	e.rpmMember = func(_ context.Context, name string, _ bool) (*RPMMember, error) {
 		calls.Add(1)
 		time.Sleep(20 * time.Millisecond)
 		return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}}, nil
@@ -160,7 +160,7 @@ func TestMemberRedirect(t *testing.T) {
 
 func debEngine(members map[string]*DebMember) *Engine {
 	e := fakeEngine(nil)
-	e.debMember = func(_ context.Context, name, base string) (*DebMember, error) {
+	e.debMember = func(_ context.Context, name, base string, _ bool) (*DebMember, error) {
 		if m := members[name]; m != nil && base == "" {
 			return m, nil
 		}
@@ -213,7 +213,7 @@ func TestFetchDebMergeIsSharedPerSuite(t *testing.T) {
 	var calls atomic.Int32
 	e := fakeEngine(nil)
 	e.mergeTTL = time.Minute
-	e.debMember = func(_ context.Context, name, base string) (*DebMember, error) {
+	e.debMember = func(_ context.Context, name, base string, _ bool) (*DebMember, error) {
 		calls.Add(1)
 		m := flatDebMember(name, "", debStanza("foo", "1", "amd64", "pool/foo.deb"))
 		return &m, nil
@@ -307,7 +307,9 @@ func suiteDebMember(name, version string) *DebMember {
 func TestFetchDebByHashSurvivesRemerge(t *testing.T) {
 	version := "1"
 	e := fakeEngine(nil)
-	e.debMember = func(_ context.Context, name, _ string) (*DebMember, error) { return suiteDebMember(name, version), nil }
+	e.debMember = func(_ context.Context, name, _ string, _ bool) (*DebMember, error) {
+		return suiteDebMember(name, version), nil
+	}
 	virt := debVirt("a")
 	fetch := func(p string) []byte {
 		body, _, err := e.Fetch(context.Background(), virt, p, "")
@@ -340,7 +342,9 @@ func TestFetchDebByHashSurvivesRemerge(t *testing.T) {
 func TestFetchDebBoundsSuitesPerVirtual(t *testing.T) {
 	e := fakeEngine(nil)
 	e.mergeTTL = time.Minute
-	e.debMember = func(_ context.Context, name, _ string) (*DebMember, error) { return suiteDebMember(name, "1"), nil }
+	e.debMember = func(_ context.Context, name, _ string, _ bool) (*DebMember, error) {
+		return suiteDebMember(name, "1"), nil
+	}
 	other := models.Virtual{Name: "other", PackageType: models.PackageDeb, Members: []string{"a"}}
 	if _, _, err := e.Fetch(context.Background(), other, "dists/keep/Release", ""); err != nil {
 		t.Fatal(err)
@@ -367,3 +371,203 @@ func TestFetchDebBoundsSuitesPerVirtual(t *testing.T) {
 		t.Fatal("another virtual's suites must not be evicted")
 	}
 }
+
+func fetchFilesInput(n int) map[string]string {
+	files := map[string]string{}
+	for i := range n {
+		files[fmt.Sprintf("k%d", i)] = fmt.Sprintf("f%d", i)
+	}
+	return files
+}
+
+func TestFetchFilesParallelRunsConcurrently(t *testing.T) {
+	var inFlight, peak atomic.Int32
+	release := make(chan struct{})
+	go func() {
+		for peak.Load() < maxFileFetches {
+			time.Sleep(time.Millisecond)
+		}
+		close(release)
+	}()
+	got, err := fetchFiles(context.Background(), true, fetchFilesInput(6), func(_ context.Context, f string) ([]byte, error) {
+		n := inFlight.Add(1)
+		defer inFlight.Add(-1)
+		for p := peak.Load(); n > p && !peak.CompareAndSwap(p, n); p = peak.Load() {
+		}
+		select {
+		case <-release:
+		case <-time.After(5 * time.Second):
+			return nil, errors.New("fetches never overlapped")
+		}
+		return []byte(f), nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if peak.Load() != maxFileFetches {
+		t.Fatalf("peak concurrency %d, want %d", peak.Load(), maxFileFetches)
+	}
+	for k, f := range fetchFilesInput(6) {
+		if string(got[k]) != f {
+			t.Fatalf("%s = %q, want %q", k, got[k], f)
+		}
+	}
+}
+
+func TestFetchFilesSerialByDefault(t *testing.T) {
+	var inFlight atomic.Int32
+	_, err := fetchFiles(context.Background(), false, fetchFilesInput(4), func(_ context.Context, f string) ([]byte, error) {
+		defer inFlight.Add(-1)
+		if inFlight.Add(1) > 1 {
+			return nil, errors.New("concurrent fetch")
+		}
+		time.Sleep(time.Millisecond)
+		return []byte(f), nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestFetchFilesErrorFailsMember(t *testing.T) {
+	boom := errors.New("boom")
+	for _, parallel := range []bool{false, true} {
+		got, err := fetchFiles(context.Background(), parallel, fetchFilesInput(5), func(_ context.Context, f string) ([]byte, error) {
+			if f == "f2" {
+				return nil, boom
+			}
+			return []byte(f), nil
+		})
+		if !errors.Is(err, boom) || got != nil {
+			t.Fatalf("parallel=%v: got %v, %v; want boom and no files", parallel, got, err)
+		}
+	}
+}
+
+func TestMergeRPMPassesParallelFetchAndKeepsMemberOrder(t *testing.T) {
+	e := fakeEngine(nil)
+	var sawParallel atomic.Bool
+	e.rpmMember = func(_ context.Context, name string, parallel bool) (*RPMMember, error) {
+		sawParallel.Store(parallel)
+		if name == "a" {
+			time.Sleep(20 * time.Millisecond)
+		}
+		return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", name+"sum", name+".rpm"))}}, nil
+	}
+	virt := rpmVirt("a", "b")
+	virt.ParallelFetch = true
+	repo, err := e.mergeRPM(context.Background(), virt)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !sawParallel.Load() {
+		t.Fatal("parallel_fetch not passed to member fetch")
+	}
+	primary := regexp.MustCompile(`repodata/[0-9a-f]+-primary\.xml\.gz`).FindString(string(repo.Repomd))
+	body := string(gunzip(t, repo.Files[primary]))
+	if a, b := strings.Index(body, "a.rpm"), strings.Index(body, "b.rpm"); a < 0 || b >= 0 && b < a {
+		t.Fatalf("member priority order lost:\n%s", body)
+	}
+}
+
+func TestFetchFilesErrorCancelsRest(t *testing.T) {
+	boom := errors.New("boom")
+	var calls atomic.Int32
+	_, err := fetchFiles(context.Background(), true, fetchFilesInput(8), func(ctx context.Context, _ string) ([]byte, error) {
+		if calls.Add(1) == 1 {
+			return nil, boom
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(5 * time.Second):
+			return nil, errors.New("in-flight fetch not cancelled")
+		}
+	})
+	if !errors.Is(err, boom) {
+		t.Fatalf("got %v, want boom", err)
+	}
+	if n := calls.Load(); n > maxFileFetches {
+		t.Fatalf("%d fetches started after the first error, want at most %d", n, maxFileFetches)
+	}
+}
+
+// overlapEngine serves fixed member index paths and records the peak number of
+// index file fetches in flight. Each fetch blocks until want are in flight (or
+// fails after 5s); with want 1 it also holds briefly so a stray overlap shows.
+func overlapEngine(index map[string][]byte, want int32, peak *atomic.Int32) *Engine {
+	e := fakeEngine(nil)
+	var inFlight atomic.Int32
+	e.memberPath = func(_ context.Context, _ models.Remote, path string) ([]byte, error) {
+		body, ok := index[path]
+		if !ok {
+			return nil, fmt.Errorf("%s: %w", path, ErrNotFound)
+		}
+		if strings.HasSuffix(path, "Release") || strings.HasSuffix(path, "repomd.xml") {
+			return body, nil
+		}
+		n := inFlight.Add(1)
+		defer inFlight.Add(-1)
+		for p := peak.Load(); n > p && !peak.CompareAndSwap(p, n); p = peak.Load() {
+		}
+		for deadline := time.Now().Add(5 * time.Second); peak.Load() < want; {
+			if time.Now().After(deadline) {
+				return nil, fmt.Errorf("%s: only %d of %d fetches in flight", path, peak.Load(), want)
+			}
+			time.Sleep(time.Millisecond)
+		}
+		if want == 1 {
+			time.Sleep(10 * time.Millisecond)
+		}
+		return body, nil
+	}
+	return e
+}
+
+func TestFetchRPMMemberHonoursParallelFetch(t *testing.T) {
+	index := map[string][]byte{
+		"repodata/repomd.xml": []byte(`<repomd>` +
+			`<data type="primary"><location href="repodata/p.xml"/></data>` +
+			`<data type="filelists"><location href="repodata/f.xml"/></data>` +
+			`<data type="other"><location href="repodata/o.xml"/></data></repomd>`),
+		"repodata/p.xml": []byte("P"), "repodata/f.xml": []byte("F"), "repodata/o.xml": []byte("O"),
+	}
+	for parallel, want := range map[bool]int32{false: 1, true: 3} {
+		var peak atomic.Int32
+		m, err := overlapEngine(index, want, &peak).fetchRPMMember(context.Background(), "a", parallel)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if peak.Load() != want {
+			t.Errorf("parallel=%v: peak concurrency %d, want %d", parallel, peak.Load(), want)
+		}
+		if string(m.Data["primary"]) != "P" || string(m.Data["filelists"]) != "F" || string(m.Data["other"]) != "O" {
+			t.Errorf("parallel=%v: data %q", parallel, m.Data)
+		}
+	}
+}
+
+func TestFetchDebMemberHonoursParallelFetch(t *testing.T) {
+	index := map[string][]byte{
+		"dists/s/Release": []byte("Suite: s\nSHA256:\n" +
+			" aa 1 main/binary-amd64/Packages\n" +
+			" bb 1 contrib/binary-amd64/Packages\n" +
+			" cc 1 non-free/binary-amd64/Packages\n"),
+		"dists/s/main/binary-amd64/Packages":     []byte("M"),
+		"dists/s/contrib/binary-amd64/Packages":  []byte("C"),
+		"dists/s/non-free/binary-amd64/Packages": []byte("N"),
+	}
+	for parallel, want := range map[bool]int32{false: 1, true: 3} {
+		var peak atomic.Int32
+		m, err := overlapEngine(index, want, &peak).fetchDebMember(context.Background(), "a", "dists/s/", parallel)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if peak.Load() != want {
+			t.Errorf("parallel=%v: peak concurrency %d, want %d", parallel, peak.Load(), want)
+		}
+		if string(m.Packages["main/binary-amd64/"]) != "M" || string(m.Packages["non-free/binary-amd64/"]) != "N" {
+			t.Errorf("parallel=%v: packages %q", parallel, m.Packages)
+		}
+	}
+}