diff --git a/internal/api/v2/virtuals_test.go b/internal/api/v2/virtuals_test.go new file mode 100644 index 0000000..2513556 --- /dev/null +++ b/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) + } +} diff --git a/internal/database/database_test.go b/internal/database/database_test.go index ef0e8a9..c97f314 100644 --- a/internal/database/database_test.go +++ b/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{ diff --git a/internal/database/virtuals.go b/internal/database/virtuals.go index ffe815c..11d6a06 100644 --- a/internal/database/virtuals.go +++ b/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 } diff --git a/internal/virtual/engine.go b/internal/virtual/engine.go index 9ab1905..a47b1dc 100644 --- a/internal/virtual/engine.go +++ b/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 diff --git a/internal/virtual/engine_test.go b/internal/virtual/engine_test.go index 04a6a82..abd22ff 100644 --- a/internal/virtual/engine_test.go +++ b/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(`` + + `` + + `` + + ``), + "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) + } + } +} diff --git a/migrations/0007_virtual_parallel_fetch.sql b/migrations/0007_virtual_parallel_fetch.sql new file mode 100644 index 0000000..0178cb1 --- /dev/null +++ b/migrations/0007_virtual_parallel_fetch.sql @@ -0,0 +1 @@ +ALTER TABLE virtuals ADD COLUMN IF NOT EXISTS parallel_fetch BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/pkg/models/virtual.go b/pkg/models/virtual.go index c7be7b2..83a39fa 100644 --- a/pkg/models/virtual.go +++ b/pkg/models/virtual.go @@ -3,11 +3,12 @@ package models import "time" type Virtual struct { - Name string `json:"name"` - PackageType PackageType `json:"package_type"` - Description string `json:"description,omitempty"` - Members []string `json:"members"` - ManagedBy string `json:"managed_by,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + Name string `json:"name"` + PackageType PackageType `json:"package_type"` + Description string `json:"description,omitempty"` + Members []string `json:"members"` + ManagedBy string `json:"managed_by,omitempty"` + ParallelFetch bool `json:"parallel_fetch"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` }