diff --git a/internal/boxcli/featureflag/feature.go b/internal/boxcli/featureflag/feature.go index 96ea8d05067..e08fcb61af2 100644 --- a/internal/boxcli/featureflag/feature.go +++ b/internal/boxcli/featureflag/feature.go @@ -7,6 +7,7 @@ import ( "log/slog" "os" "strconv" + "sync" "testing" "go.jetify.com/devbox/internal/build" @@ -43,7 +44,9 @@ func enable(name string) *feature { return features[name] } -var logMap = map[string]bool{} +// logged records which env var overrides have been logged. Enabled can be +// called from multiple goroutines. +var logged sync.Map func (f *feature) Enabled() bool { if f == nil { @@ -54,9 +57,8 @@ func (f *feature) Enabled() bool { if !on { status = "disabled" } - if !logMap[f.name] { + if _, loaded := logged.LoadOrStore(f.name, true); !loaded { slog.Debug("feature flag set via environment variable", "feature", f.name, "status", status) - logMap[f.name] = true } return on } diff --git a/internal/boxcli/multi/sync.go b/internal/boxcli/multi/sync.go index c13fe28d849..9243171fdad 100644 --- a/internal/boxcli/multi/sync.go +++ b/internal/boxcli/multi/sync.go @@ -55,7 +55,7 @@ func SyncLockfiles(pkgs []string) error { } if changed { - if err = cuecfg.WriteFile(lockfilePath, lockFile); err != nil { + if err = cuecfg.WriteFile(lockfilePath, &lockFile); err != nil { return err } fmt.Printf("Updated: %s\n", lockfilePath) diff --git a/internal/boxcli/multi/sync_test.go b/internal/boxcli/multi/sync_test.go index 94ca961a742..8ca06b7f19e 100644 --- a/internal/boxcli/multi/sync_test.go +++ b/internal/boxcli/multi/sync_test.go @@ -27,7 +27,7 @@ func TestSyncLockfilesOnlyRequestedPackages(t *testing.T) { f.Packages[key] = &lock.Package{LastModified: lastModified, Resolved: key + "-" + lastModified} } path := filepath.Join(dir, "devbox.lock") - if err := cuecfg.WriteFile(path, f); err != nil { + if err := cuecfg.WriteFile(path, &f); err != nil { t.Fatal(err) } return path diff --git a/internal/devpkg/narinfo_cache.go b/internal/devpkg/narinfo_cache.go index 0635ef1453a..d79d40c4964 100644 --- a/internal/devpkg/narinfo_cache.go +++ b/internal/devpkg/narinfo_cache.go @@ -54,16 +54,7 @@ func (p *Package) IsInBinaryCache() (bool, error) { func FillNarInfoCache(ctx context.Context, packages ...*Package) error { defer debug.FunctionTimer().End() - eligiblePackages := []*Package{} - for _, p := range packages { - // IMPORTANT: isEligibleForBinaryCache will call resolve() which is NOT - // concurrency safe. Hence, we call it outside of the go-routine. - isEligible, err := p.isEligibleForBinaryCache() - // If the package is not eligible or there is an error in determining that, then skip it. - if isEligible && err == nil { - eligiblePackages = append(eligiblePackages, p) - } - } + eligiblePackages := eligibleForBinaryCache(packages) if len(eligiblePackages) == 0 { return nil } @@ -87,6 +78,37 @@ func FillNarInfoCache(ctx context.Context, packages ...*Package) error { return group.Wait() } +// maxConcurrentResolves bounds how many packages are resolved at once. Each +// resolve can make a search API request or run `nix flake prefetch`. +const maxConcurrentResolves = 8 + +// eligibleForBinaryCache returns the packages that are eligible for the binary +// cache, preserving their order. Checking eligibility resolves packages that +// aren't in the lockfile yet, so the checks run concurrently. +func eligibleForBinaryCache(packages []*Package) []*Package { + eligible := make([]bool, len(packages)) + var group errgroup.Group + group.SetLimit(maxConcurrentResolves) + for i, p := range packages { + group.Go(func() error { + // If the package is not eligible or there is an error in + // determining that, then skip it. + isEligible, err := p.isEligibleForBinaryCache() + eligible[i] = isEligible && err == nil + return nil + }) + } + _ = group.Wait() + + var result []*Package + for i, p := range packages { + if eligible[i] { + result = append(result, p) + } + } + return result +} + // areExpectedOutputsInCacheOnce wraps fetchNarInfoStatusOnce and returns true // if the expected outputs are in the cache. func (p *Package) areExpectedOutputsInCacheOnce(outputName string) (bool, error) { @@ -177,8 +199,8 @@ func (p *Package) isEligibleForBinaryCache() (bool, error) { // sysInfoIfExists returns the system info for the user's system. If the sysInfo // is missing, then nil is returned // NOTE: this is called from multiple go-routines and needs to be concurrency safe. -// Hence, we compute nix.Version, nix.System and lockfile.Resolve prior to calling this -// function from within a goroutine. +// nix.Version and nix.System are computed once, and lock.File.Resolve is safe to +// call concurrently. func (p *Package) sysInfoIfExists() (*lock.SystemInfo, error) { if !p.isVersioned() { return nil, nil diff --git a/internal/devpkg/narinfo_cache_test.go b/internal/devpkg/narinfo_cache_test.go index 2850c5dd419..7b5ce714396 100644 --- a/internal/devpkg/narinfo_cache_test.go +++ b/internal/devpkg/narinfo_cache_test.go @@ -8,10 +8,15 @@ import ( "net/http/httptest" "os" "path/filepath" + "slices" + "strings" + "sync" "sync/atomic" "testing" "time" + "go.jetify.com/devbox/internal/lock" + "go.jetify.com/devbox/internal/nix" "go.jetify.com/pkg/filecache" ) @@ -242,3 +247,70 @@ func TestFetchNarInfoStatusCache(t *testing.T) { } }) } + +// blockingLocker is a lock.Locker whose Resolve blocks until release is +// closed, and reports each call on inFlight. +type blockingLocker struct { + lockfile + inFlight chan struct{} + release chan struct{} +} + +func (l *blockingLocker) Resolve(pkg string) (*lock.Package, error) { + l.inFlight <- struct{}{} + <-l.release + + entry := &lock.Package{ + Resolved: "github:NixOS/nixpkgs/" + nixCommitHash + "#" + pkg, + } + // Packages named "uncached@..." have no system info, so they're not + // eligible for the binary cache. + if !strings.HasPrefix(pkg, "uncached@") { + entry.Systems = map[string]*lock.SystemInfo{nix.System(): {}} + } + return entry, nil +} + +func TestEligibleForBinaryCacheResolvesConcurrently(t *testing.T) { + if !nix.AtLeast(nix.Version2_17) { + t.Skip("binary cache eligibility requires nix >= 2.17") + } + + raws := []string{"hello@1", "uncached@1", "jq@1", "curl@1"} + locker := &blockingLocker{ + inFlight: make(chan struct{}, len(raws)), + release: make(chan struct{}), + } + unblock := sync.OnceFunc(func() { close(locker.release) }) + // Don't leave resolves blocked if the test fails early. + t.Cleanup(unblock) + pkgs := make([]*Package, len(raws)) + for i, raw := range raws { + pkgs[i] = PackageFromStringWithDefaults(raw, locker) + } + + done := make(chan []*Package) + go func() { done <- eligibleForBinaryCache(pkgs) }() + + // All resolves must be in flight at once. If eligibility were checked + // serially, only the first would start and this would time out. + timeout := time.After(10 * time.Second) + for range raws { + select { + case <-locker.inFlight: + case <-timeout: + t.Fatal("packages were not resolved concurrently") + } + } + unblock() + + got := <-done + var gotRaws []string + for _, p := range got { + gotRaws = append(gotRaws, p.Raw) + } + want := []string{"hello@1", "jq@1", "curl@1"} + if !slices.Equal(gotRaws, want) { + t.Errorf("got eligible packages %v, want %v", gotRaws, want) + } +} diff --git a/internal/lock/lockfile.go b/internal/lock/lockfile.go index caafc9e86d6..f16a1ca7c8e 100644 --- a/internal/lock/lockfile.go +++ b/internal/lock/lockfile.go @@ -10,6 +10,7 @@ import ( "path/filepath" "slices" "strings" + "sync" "github.com/pkg/errors" "go.jetify.com/devbox/internal/cachehash" @@ -18,6 +19,7 @@ import ( "go.jetify.com/devbox/internal/searcher" "go.jetify.com/devbox/nix/flake" "go.jetify.com/pkg/runx/impl/types" + "golang.org/x/sync/singleflight" "go.jetify.com/devbox/internal/cuecfg" ) @@ -25,9 +27,18 @@ import ( const lockFileVersion = "1" // Lightly inspired by package-lock.json +// +// File's methods are safe to call concurrently. Code that reads or writes +// Packages directly must not run while other goroutines call File methods. type File struct { devboxProject `json:"-"` + // mu guards Packages and the entries in it. It is never held while + // resolving a package, which can make network requests or run nix. + mu sync.Mutex + // resolving dedupes concurrent resolutions of the same package. + resolving singleflight.Group + LockFileVersion string `json:"lockfile_version"` // Packages is keyed by "canonicalName@version" @@ -65,20 +76,59 @@ func (f *File) Add(pkgs ...string) error { } func (f *File) Remove(pkgs ...string) error { + f.mu.Lock() for _, p := range pkgs { delete(f.Packages, p) } + f.mu.Unlock() return f.Save() } // Resolve updates the in memory copy for performance but does not write to disk // This avoids writing values that may need to be removed in case of error. +// +// Resolve is safe to call concurrently. The lockfile's lock is only held to read +// and write the entry; the search or flake request runs outside of it. +// Concurrent calls for the same package share a single request. func (f *File) Resolve(pkg string) (*Package, error) { - entry, hasEntry := f.Packages[pkg] - if hasEntry && entry.Resolved != "" { - return f.Packages[pkg], nil + if entry := f.resolvedEntry(pkg); entry != nil { + return entry, nil + } + result, err, _ := f.resolving.Do(pkg, func() (any, error) { + // Another call may have finished resolving pkg between the check + // above and joining this group. + if entry := f.resolvedEntry(pkg); entry != nil { + return entry, nil + } + locked, err := f.fetchLockEntry(pkg) + if err != nil { + return nil, err + } + f.mu.Lock() + defer f.mu.Unlock() + f.Packages[pkg] = locked + return locked, nil + }) + if err != nil { + return nil, err } + return result.(*Package), nil +} + +// resolvedEntry returns the lockfile entry for pkg if it's already resolved, or +// nil otherwise. +func (f *File) resolvedEntry(pkg string) *Package { + f.mu.Lock() + defer f.mu.Unlock() + if entry, ok := f.Packages[pkg]; ok && entry.Resolved != "" { + return entry + } + return nil +} +// fetchLockEntry computes a new lockfile entry for pkg without storing it. It +// must not be called with f.mu held. +func (f *File) fetchLockEntry(pkg string) (*Package, error) { locked := &Package{} _, _, versioned := searcher.ParseVersionedPackage(pkg) if pkgtype.IsRunX(pkg) || versioned || pkgtype.IsFlake(pkg) { @@ -100,9 +150,7 @@ func (f *File) Resolve(pkg string) (*Package, error) { Source: nixpkgSource, } } - f.Packages[pkg] = locked - - return f.Packages[pkg], nil + return locked, nil } // TODO: @@ -116,7 +164,10 @@ func (f *File) Resolve(pkg string) (*Package, error) { // 2. Then, in Save(), we can check if OutputsRaw is zero and fill it in prior to writing // to disk. func (f *File) Save() error { - isDirty, err := f.isDirty() + f.mu.Lock() + defer f.mu.Unlock() + + isDirty, err := f.isDirtyLocked() if err != nil { return err } @@ -148,11 +199,15 @@ func (f *File) UpdateStdenv() error { // Remove the old entry in memory only so that a failed resolve doesn't // leave the lockfile without a stdenv. stdenv := f.devboxProject.Stdenv().String() + f.mu.Lock() old, hasOld := f.Packages[stdenv] delete(f.Packages, stdenv) + f.mu.Unlock() if err := f.Add(stdenv); err != nil { if hasOld { + f.mu.Lock() f.Packages[stdenv] = old + f.mu.Unlock() } return err } @@ -180,14 +235,12 @@ func (f *File) Stdenv() flake.Ref { } func (f *File) Get(pkg string) *Package { - entry, hasEntry := f.Packages[pkg] - if !hasEntry || entry.Resolved == "" { - return nil - } - return entry + return f.resolvedEntry(pkg) } func (f *File) HasAllowInsecurePackages() bool { + f.mu.Lock() + defer f.mu.Unlock() for _, pkg := range f.Packages { if pkg.AllowInsecure { return true @@ -214,6 +267,8 @@ func IsLegacyPackage(pkg string) bool { func (f *File) Tidy() { keep := f.devboxProject.AllPackageNamesIncludingRemovedTriggerPackages() keep = append(keep, f.devboxProject.Stdenv().String()) + f.mu.Lock() + defer f.mu.Unlock() maps.DeleteFunc(f.Packages, func(key string, pkg *Package) bool { return !slices.Contains(keep, key) }) @@ -253,21 +308,29 @@ func (f *File) StateStatus(isFish bool) (StateStatus, error) { } func (f *File) SetOutputsForPackage(pkg string, outputs []Output) error { - p, err := f.Resolve(pkg) + entry, err := f.Resolve(pkg) if err != nil { return err } - if p.Systems == nil { - p.Systems = map[string]*SystemInfo{} + f.mu.Lock() + if entry.Systems == nil { + entry.Systems = map[string]*SystemInfo{} } - if p.Systems[nix.System()] == nil { - p.Systems[nix.System()] = &SystemInfo{} + if entry.Systems[nix.System()] == nil { + entry.Systems[nix.System()] = &SystemInfo{} } - p.Systems[nix.System()].Outputs = outputs + entry.Systems[nix.System()].Outputs = outputs + f.mu.Unlock() return f.Save() } func (f *File) isDirty() (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.isDirtyLocked() +} + +func (f *File) isDirtyLocked() (bool, error) { currentHash, err := cachehash.JSON(f) if err != nil { return false, err diff --git a/internal/lock/lockfile_test.go b/internal/lock/lockfile_test.go new file mode 100644 index 00000000000..ebe76d7c294 --- /dev/null +++ b/internal/lock/lockfile_test.go @@ -0,0 +1,212 @@ +// Copyright 2024 Jetify Inc. and contributors. All rights reserved. +// Use of this source code is governed by the license in the LICENSE file. + +package lock + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "go.jetify.com/devbox/internal/envir" + "go.jetify.com/devbox/internal/nix" + "go.jetify.com/devbox/internal/searcher" +) + +// fakeResolveServer serves /v1/resolve. It counts requests per package name, +// and its handler blocks until unblock is called, so tests can control when +// requests finish. +type fakeResolveServer struct { + *httptest.Server + + mu sync.Mutex + requests map[string]int + inFlight chan string + release chan struct{} + // unblock releases blocked handlers. It's safe to call more than once. + unblock func() +} + +func newFakeResolveServer(t *testing.T) *fakeResolveServer { + t.Helper() + srv := &fakeResolveServer{ + requests: map[string]int{}, + inFlight: make(chan string, 100), + release: make(chan struct{}), + } + srv.unblock = sync.OnceFunc(func() { close(srv.release) }) + srv.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + name := r.URL.Query().Get("name") + srv.mu.Lock() + srv.requests[name]++ + srv.mu.Unlock() + srv.inFlight <- name + <-srv.release + + info := searcher.PackageInfo{ + CommitHash: "1111111111111111111111111111111111111111", + AttrPaths: []string{name}, + Version: r.URL.Query().Get("version"), + } + _ = json.NewEncoder(w).Encode(searcher.PackageVersion{ + PackageInfo: info, + Name: name, + // No store hash, so resolving doesn't shell out to nix. + Systems: map[string]searcher.PackageInfo{"x86_64-linux": info}, + }) + })) + t.Cleanup(srv.Close) + // Cleanups run in reverse order, so this releases any handlers still + // blocked (e.g. after a t.Fatal) before Close waits for them. + t.Cleanup(srv.unblock) + t.Setenv(envir.DevboxSearchHost, srv.URL) + t.Setenv("DEVBOX_FEATURE_RESOLVE_V2", "0") + return srv +} + +func (s *fakeResolveServer) requestCount(name string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.requests[name] +} + +// TestResolveConcurrent checks that concurrent Resolve calls for different +// packages make their search requests in parallel, that concurrent calls for +// the same package share one request, and that every caller gets the stored +// entry. +func TestResolveConcurrent(t *testing.T) { + srv := newFakeResolveServer(t) + lockfile := &File{Packages: map[string]*Package{}} + + names := []string{"hello", "jq", "ripgrep", "curl", "git"} + const callersPerPkg = 4 + + type result struct { + pkg string + entry *Package + err error + } + results := make(chan result, len(names)*callersPerPkg) + var callers sync.WaitGroup + for _, name := range names { + pkg := name + "@1.0" + for range callersPerPkg { + callers.Go(func() { + entry, err := lockfile.Resolve(pkg) + results <- result{pkg, entry, err} + }) + } + } + + // Every package's request must be in flight at the same time. If Resolve + // held the lockfile lock during the request, only one would arrive and + // this would time out. + seen := map[string]bool{} + timeout := time.After(10 * time.Second) + for len(seen) < len(names) { + select { + case name := <-srv.inFlight: + seen[name] = true + case <-timeout: + t.Fatalf("got %d concurrent requests, want %d: requests are serialized", len(seen), len(names)) + } + } + + // While the requests are blocked, the lockfile must stay usable. + if lockfile.Get("hello@1.0") != nil { + t.Error("Get returned an entry for a package that is still resolving") + } + _ = lockfile.HasAllowInsecurePackages() + + srv.unblock() + callers.Wait() + close(results) + + for r := range results { + if r.err != nil { + t.Fatalf("Resolve(%q): %v", r.pkg, r.err) + } + if got := lockfile.Get(r.pkg); got != r.entry { + t.Errorf("Resolve(%q) returned %p, but the lockfile has %p", r.pkg, r.entry, got) + } + name := r.pkg[:len(r.pkg)-len("@1.0")] + want := fmt.Sprintf("github:NixOS/nixpkgs/1111111111111111111111111111111111111111#%s", name) + if r.entry.Resolved != want { + t.Errorf("Resolve(%q).Resolved = %q, want %q", r.pkg, r.entry.Resolved, want) + } + } + for _, name := range names { + if got := srv.requestCount(name); got != 1 { + t.Errorf("got %d search requests for %q, want 1", got, name) + } + } + + // Resolving again uses the lockfile and makes no requests. + if _, err := lockfile.Resolve("hello@1.0"); err != nil { + t.Fatal(err) + } + if got := srv.requestCount("hello"); got != 1 { + t.Errorf("got %d search requests for hello after re-resolving, want 1", got) + } +} + +// TestResolveConcurrentError checks that a failed resolve is returned to every +// concurrent caller, isn't stored, and is retried by the next call. +func TestResolveConcurrentError(t *testing.T) { + var fail atomic.Bool + fail.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if fail.Load() { + w.WriteHeader(http.StatusNotFound) + return + } + info := searcher.PackageInfo{ + CommitHash: "1111111111111111111111111111111111111111", + AttrPaths: []string{"hello"}, + Version: "1.0", + } + _ = json.NewEncoder(w).Encode(searcher.PackageVersion{ + PackageInfo: info, + Name: "hello", + Systems: map[string]searcher.PackageInfo{"x86_64-linux": info}, + }) + })) + t.Cleanup(srv.Close) + t.Setenv(envir.DevboxSearchHost, srv.URL) + t.Setenv("DEVBOX_FEATURE_RESOLVE_V2", "0") + + lockfile := &File{Packages: map[string]*Package{}} + var callers sync.WaitGroup + errs := make(chan error, 8) + for range 8 { + callers.Go(func() { + _, err := lockfile.Resolve("hello@1.0") + errs <- err + }) + } + callers.Wait() + close(errs) + for err := range errs { + if !errors.Is(err, nix.ErrPackageNotFound) { + t.Errorf("got error %v, want nix.ErrPackageNotFound", err) + } + } + if lockfile.Get("hello@1.0") != nil { + t.Error("failed resolve was stored in the lockfile") + } + + fail.Store(false) + entry, err := lockfile.Resolve("hello@1.0") + if err != nil { + t.Fatalf("Resolve after failure: %v", err) + } + if lockfile.Get("hello@1.0") != entry { + t.Error("successful resolve wasn't stored in the lockfile") + } +} diff --git a/specs/perf-improvements.md b/specs/perf-improvements.md index 939a43e944d..65183f49e2b 100644 --- a/specs/perf-improvements.md +++ b/specs/perf-improvements.md @@ -215,7 +215,7 @@ serially. Fix: resolve in an `errgroup` with bounded concurrency, then apply lockfile writes serially. Preserve the existing error semantics (flakes warn-and-continue, versioned packages abort). -### 13. Resolve lockfile-missing packages in parallel +### 13. Resolve lockfile-missing packages in parallel (done in #3054) **Effort:** ~60–100 LOC. diff --git a/vendor-hash b/vendor-hash index aabc3eefe80..ead41b39363 100644 --- a/vendor-hash +++ b/vendor-hash @@ -1 +1 @@ -sha256-HSWDqEGmqZtkQHw96lYewQFDBWt7zbV3xtFXyf07llM= +sha256-YrE6OR+21Nwf6EFAJkZk7HSN3t+JzA4TgRNH+ziQYYQ=