From 669b5f40f91d894a9da98b59a759f43b2fcded52 Mon Sep 17 00:00:00 2001 From: montehurd Date: Mon, 7 Sep 2026 12:38:24 -0700 Subject: [PATCH 1/2] Return the stored artifact from storeArtifact, not a reader storeArtifact returned a CacheResult holding an open file handle. A handle has one read position, so it can only ever serve a single caller, which is what blocks sharing one fetch between concurrent requests. Return the artifact and its storage path instead, and let each caller open its own reader through openStoredArtifact. Threading that type through fetchAndCache, fetchAndCacheFromURL and their error paths is mechanical; behaviour is unchanged. --- internal/handler/handler.go | 49 +++++++++++++++++++++++++------------ 1 file changed, 33 insertions(+), 16 deletions(-) diff --git a/internal/handler/handler.go b/internal/handler/handler.go index a78393d..856e673 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -225,7 +225,11 @@ func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version } metrics.RecordCacheMiss(ecosystem) - return p.fetchAndCache(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL) + stored, storagePath, err := p.fetchAndCache(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL) + if err != nil { + return nil, err + } + return p.openStoredArtifact(ctx, stored, storagePath) } // GetCachedArtifact retrieves an artifact from cache without contacting an upstream. @@ -359,14 +363,14 @@ func (p *Proxy) rejectUnusableCacheRecord(artifact *database.CachedArtifact, ver } } -func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL string) (*CacheResult, error) { +func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL string) (artifacts.Artifact, string, error) { // Resolve download URL info, err := p.Resolver.Resolve(ctx, ecosystem, name, version) if err != nil { if errors.Is(err, fetch.ErrNotFound) { - return nil, ErrUpstreamNotFound + return artifacts.Artifact{}, "", ErrUpstreamNotFound } - return nil, fmt.Errorf("resolving download URL: %w", err) + return artifacts.Artifact{}, "", fmt.Errorf("resolving download URL: %w", err) } // Use resolved filename if provided filename is empty @@ -386,9 +390,9 @@ func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, fil metrics.RecordUpstreamFetch(ecosystem, fetchDuration) metrics.RecordUpstreamError(ecosystem, "fetch_failed") if errors.Is(err, fetch.ErrNotFound) { - return nil, ErrUpstreamNotFound + return artifacts.Artifact{}, "", ErrUpstreamNotFound } - return nil, fmt.Errorf("fetching from upstream: %w", err) + return artifacts.Artifact{}, "", fmt.Errorf("fetching from upstream: %w", err) } metrics.RecordUpstreamFetch(ecosystem, fetchDuration) @@ -405,7 +409,10 @@ func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, fil // verdict means a blocked artifact was never reachable by any client. On // block, the just-stored bytes are deleted and ErrArtifactBlocked is // returned; updateCacheDB is never called. -func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash string, artifact *fetch.Artifact) (*CacheResult, error) { +// +// It returns the artifact and its storage path, not a reader; callers get one +// from openStoredArtifact. +func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash string, artifact *fetch.Artifact) (artifacts.Artifact, string, error) { storagePath := storage.ArtifactPath(ecosystem, "", name, version, filename) storeStart := time.Now() @@ -414,14 +421,14 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil metrics.RecordStorageOperation("write", time.Since(storeStart)) if err != nil { metrics.RecordStorageError("write") - return nil, fmt.Errorf("storing artifact: %w", err) + return artifacts.Artifact{}, "", fmt.Errorf("storing artifact: %w", err) } if !artifactHashMatches(hash, upstreamHash) { if delErr := p.Storage.Delete(ctx, storagePath); delErr != nil { p.Logger.Warn("failed to discard artifact with mismatched checksum", "path", storagePath, "error", delErr) } - return nil, fmt.Errorf("%w: upstream declared %s, got %s", ErrArtifactDigestMismatch, upstreamHash, hash) + return artifacts.Artifact{}, "", fmt.Errorf("%w: upstream declared %s, got %s", ErrArtifactDigestMismatch, upstreamHash, hash) } if p.Scanners != nil && p.Scanners.Enabled() { @@ -433,7 +440,7 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil p.Logger.Warn("failed to delete blocked artifact from storage", "path", storagePath, "error", delErr) } - return nil, err + return artifacts.Artifact{}, "", err } } @@ -451,7 +458,13 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil // Continue anyway - we have the file } - // Open the stored file to return + return sharedArtifact, storagePath, nil +} + +// openStoredArtifact gives one caller its own reader over just-committed +// bytes. A handle cannot be shared: it has one read position, so callers would +// consume each other's bytes and the first Close would break the rest. +func (p *Proxy) openStoredArtifact(ctx context.Context, artifact artifacts.Artifact, storagePath string) (*CacheResult, error) { readStart := time.Now() reader, err := p.Storage.Open(ctx, storagePath) metrics.RecordStorageOperation("read", time.Since(readStart)) @@ -463,7 +476,7 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil return &CacheResult{ Reader: reader, - Artifact: sharedArtifact, + Artifact: artifact, Cached: false, }, nil } @@ -1101,7 +1114,11 @@ func (p *Proxy) getOrFetchArtifactFromURLWithCachePURLs(ctx context.Context, eco } metrics.RecordCacheMiss(ecosystem) - return p.fetchAndCacheFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash) + stored, storagePath, err := p.fetchAndCacheFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash) + if err != nil { + return nil, err + } + return p.openStoredArtifact(ctx, stored, storagePath) } // getCachedArtifactWithUpstreamHash returns a cached artifact whose recorded @@ -1128,7 +1145,7 @@ func (p *Proxy) getCachedArtifactWithUpstreamHash(ctx context.Context, pkgPURL, return nil, nil } -func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL string, headers http.Header, upstreamHash string) (*CacheResult, error) { +func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL string, headers http.Header, upstreamHash string) (artifacts.Artifact, string, error) { p.Logger.Info("fetching from upstream", "ecosystem", ecosystem, "name", name, "version", version, "url", downloadURL) @@ -1138,9 +1155,9 @@ func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, versi if err != nil { metrics.RecordUpstreamError(ecosystem, "fetch_failed") if errors.Is(err, fetch.ErrNotFound) { - return nil, ErrUpstreamNotFound + return artifacts.Artifact{}, "", ErrUpstreamNotFound } - return nil, fmt.Errorf("fetching from upstream: %w", err) + return artifacts.Artifact{}, "", fmt.Errorf("fetching from upstream: %w", err) } return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, upstreamHash, artifact) From 449109466becf540d1e50b9654483eafac04d483 Mon Sep 17 00:00:00 2001 From: montehurd Date: Mon, 7 Sep 2026 12:38:24 -0700 Subject: [PATCH 2/2] Coalesce concurrent cache misses A cache miss went from checkCache straight to an upstream fetch with nothing tracking in-flight work, so N concurrent requests for one uncached artifact produced N upstream fetches and N stores to the same key. That is the CI shape: parallel jobs installing overlapping dependencies against a cold cache. The duplicate stores also fail requests, racing fileblob's per-key ".attrs" sidecar into a partial read served as a 502. Over 12 runs of 8 simultaneous requests for one uncached tarball, against bb2205a: before, 8 fetches per run and 12 of 96 responses were 502; after, 1 fetch per run and none failed. Route both miss paths through a shared in-flight map keyed on the artifact, including the download URL and upstream-declared hash so callers expecting different bytes never share a fetch. singleflight does not fit: Do gives waiters no way to leave, while DoChan lets the caller running the fetch abandon it, breaking storeArtifact's scan-on-disconnect contract. Deciding roles under a mutex gives both behaviours. The fetch runs on the first caller's context and is seen through; waiters leave when their own clients do. This removes the sidecar trigger on this path. The race is in fileblob and three writers bypass this path entirely, so it is fixed separately. Fewer failures now reach the circuit breaker, so it trips later. Sixteen concurrent callers against real file:// storage fail 10 of 10 runs on main and pass 10 of 10 here. Other tests pin key discrimination, failure propagation, resolver-path coalescing, per-caller readers, waiter cancellation, key release and panic safety. allocs/op is unchanged. mockStorage gains a mutex so concurrent tests can use it. --- internal/handler/coalesce_semantics_test.go | 442 ++++++++++++++++++++ internal/handler/coalesce_test.go | 176 ++++++++ internal/handler/handler.go | 121 +++++- internal/handler/handler_test.go | 14 + 4 files changed, 741 insertions(+), 12 deletions(-) create mode 100644 internal/handler/coalesce_semantics_test.go create mode 100644 internal/handler/coalesce_test.go diff --git a/internal/handler/coalesce_semantics_test.go b/internal/handler/coalesce_semantics_test.go new file mode 100644 index 0000000..40a2795 --- /dev/null +++ b/internal/handler/coalesce_semantics_test.go @@ -0,0 +1,442 @@ +package handler + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/git-pkgs/registries/fetch" +) + +// runConcurrent runs fn in n goroutines released together and returns their errors. +func runConcurrent(n int, fn func(i int) error) []error { + errs := make([]error, n) + start := make(chan struct{}) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + errs[i] = fn(i) + }(i) + } + close(start) + wg.Wait() + return errs +} + +// artifactBody builds a one-shot upstream artifact carrying the given bytes. +func artifactBody(content string) *fetch.Artifact { + return &fetch.Artifact{ + Body: io.NopCloser(strings.NewReader(content)), + ContentType: "application/gzip", + } +} + +// drain consumes and closes a CacheResult reader, if there is one. +func drain(res *CacheResult) { + if res != nil && res.Reader != nil { + _, _ = io.Copy(io.Discard, res.Reader) + _ = res.Reader.Close() + } +} + +// TestCoalesceKey_DifferentUpstreamHashDoesNotShare is the safety property that +// makes coalescing sound: callers expecting different bytes must never share a +// fetch, so a re-published version cannot serve stale bytes to a caller that +// asked for the new digest. +func TestCoalesceKey_DifferentUpstreamHashDoesNotShare(t *testing.T) { + const content = "artifact bytes" + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: content, delay: fetchHoldTime} + proxy.Fetcher = fetcher + + // The digest must carry the "sha256:" prefix; without it the API treats the + // value as unverifiable and clears the hash, which would legitimately let + // the two callers share one fetch. + hashes := []string{ + "sha256:" + sha256Hex(content), + "sha256:" + sha256Hex("something else entirely"), + } + + _ = runConcurrent(2, func(i int) error { + res, err := proxy.GetOrFetchArtifactFromURLWithDigest(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", + "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz", hashes[i]) + drain(res) + return err + }) + + if got := fetcher.calls.Load(); got != 2 { + t.Errorf("upstream fetches = %d, want 2: callers expecting different digests must not share a fetch", got) + } +} + +// TestCoalesceKey_DifferentDownloadURLDoesNotShare covers the other half of the +// key: same package, different upstream URL, must not collapse into one fetch. +func TestCoalesceKey_DifferentDownloadURLDoesNotShare(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "artifact bytes", delay: fetchHoldTime} + proxy.Fetcher = fetcher + + urls := []string{ + "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz", + "https://mirror.example.com/pkg/-/pkg-1.0.0.tgz", + } + + _ = runConcurrent(2, func(i int) error { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", urls[i]) + drain(res) + return err + }) + + if got := fetcher.calls.Load(); got != 2 { + t.Errorf("upstream fetches = %d, want 2: different upstream URLs must not share a fetch", got) + } +} + +// TestCoalesceKey_DistinctArtifactsDoNotSerialize guards against an over-broad +// key: four packages fetched at once must still produce four fetches. +func TestCoalesceKey_DistinctArtifactsDoNotSerialize(t *testing.T) { + const n = 4 + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "artifact bytes", delay: fetchHoldTime} + proxy.Fetcher = fetcher + + names := []string{"alpha", "beta", "gamma", "delta"} + errs := runConcurrent(n, func(i int) error { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", names[i], "1.0.0", names[i]+"-1.0.0.tgz", + "https://registry.npmjs.org/"+names[i]+"/-/"+names[i]+"-1.0.0.tgz") + drain(res) + return err + }) + for i, err := range errs { + if err != nil { + t.Errorf("caller %d (%s): %v", i, names[i], err) + } + } + if got := fetcher.calls.Load(); got != n { + t.Errorf("upstream fetches = %d, want %d: distinct artifacts must not share a fetch", got, n) + } +} + +// TestCoalesce_FailedFetchReachesEveryCallerAndIsRetriable verifies both claims +// in coalesceFetch's doc comment: a failed fetch reaches every caller sharing +// it, and the key is released so a later request retries. +func TestCoalesce_FailedFetchReachesEveryCallerAndIsRetriable(t *testing.T) { + const callers = 8 + proxy, _, _, fetcher := setupTestProxy(t) + boom := errors.New("upstream unavailable") + fetcher.fetchErr = boom + + errs := runConcurrent(callers, func(int) error { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", + "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz") + drain(res) + return err + }) + for i, err := range errs { + if err == nil { + t.Errorf("caller %d: got nil error, want the shared fetch's failure", i) + } else if !errors.Is(err, boom) { + t.Errorf("caller %d: got %v, want it to wrap %v", i, err, boom) + } + } + + // The key must be released: a later request retries rather than inheriting + // the failure. + fetcher.fetchErr = nil + fetcher.artifact = artifactBody("recovered bytes") + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", + "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz") + if err != nil { + t.Fatalf("retry after failed coalesced fetch: %v", err) + } + body, _ := io.ReadAll(res.Reader) + _ = res.Reader.Close() + if string(body) != "recovered bytes" { + t.Errorf("retry body = %q, want %q", body, "recovered bytes") + } +} + +// TestCoalesce_ResolverPath covers the other entry point: GetOrFetchArtifact +// resolves the URL itself, so it is keyed without one. +func TestCoalesce_ResolverPath(t *testing.T) { + const callers = 8 + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "resolved artifact bytes", delay: fetchHoldTime} + proxy.Fetcher = fetcher + + errs := runConcurrent(callers, func(int) error { + res, err := proxy.GetOrFetchArtifact(context.Background(), + "npm", "left-pad", "1.3.0", "left-pad-1.3.0.tgz") + drain(res) + return err + }) + for i, err := range errs { + if err != nil { + t.Errorf("caller %d: %v", i, err) + } + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1", got) + } +} + +// TestCoalesce_ResolverPathEmptyFilename exercises that path when the filename +// is left to be resolved, which the key cannot know up front. +func TestCoalesce_ResolverPathEmptyFilename(t *testing.T) { + const callers = 8 + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "resolved artifact bytes", delay: fetchHoldTime} + proxy.Fetcher = fetcher + + errs := runConcurrent(callers, func(int) error { + res, err := proxy.GetOrFetchArtifact(context.Background(), "npm", "left-pad", "1.3.0", "") + drain(res) + return err + }) + for i, err := range errs { + if err != nil { + t.Errorf("caller %d: %v", i, err) + } + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1", got) + } +} + +// TestCoalesce_SubsequentRequestIsACacheHit confirms the coalesced fetch was +// committed and is visible later, not just streamed to the waiting callers. +func TestCoalesce_SubsequentRequestIsACacheHit(t *testing.T) { + const callers = 8 + const url = "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz" + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "artifact bytes", delay: fetchHoldTime} + proxy.Fetcher = fetcher + + _ = runConcurrent(callers, func(int) error { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + drain(res) + return err + }) + + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + if err != nil { + t.Fatalf("follow-up request: %v", err) + } + defer func() { _ = res.Reader.Close() }() + if !res.Cached { + t.Error("follow-up request should be served from cache") + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1 after a follow-up cache hit", got) + } +} + +// TestCoalesce_ReadersAreIndependent guards openStoredArtifact: callers sharing +// a fetch each need their own reader, or one closing early breaks the rest. +func TestCoalesce_ReadersAreIndependent(t *testing.T) { + const callers = 8 + const content = "artifact bytes that every caller must receive intact" + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: content, delay: fetchHoldTime} + proxy.Fetcher = fetcher + + results := make([]*CacheResult, callers) + errs := runConcurrent(callers, func(i int) error { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", + "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz") + results[i] = res + return err + }) + for i, err := range errs { + if err != nil { + t.Fatalf("caller %d: %v", i, err) + } + } + + // Close the first caller's reader before anyone else has read a byte. + _ = results[0].Reader.Close() + + for i := 1; i < callers; i++ { + body, err := io.ReadAll(results[i].Reader) + _ = results[i].Reader.Close() + if err != nil { + t.Errorf("caller %d read after another caller closed: %v", i, err) + continue + } + if string(body) != content { + t.Errorf("caller %d got %q, want %q", i, body, content) + } + } +} + +// TestCoalesce_CanceledWaiterDoesNotWaitForTheSharedFetch checks that joining a +// coalesced fetch does not cost a caller its own cancellation. Without the +// leader/waiter split a waiter is pinned until the shared fetch resolves, +// bounded only by the artifact client timeout, so clients that have already +// gone away keep handler goroutines alive for minutes. +func TestCoalesce_CanceledWaiterDoesNotWaitForTheSharedFetch(t *testing.T) { + const leaderFetch = 2 * time.Second + const url = "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz" + + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "artifact bytes", delay: leaderFetch} + proxy.Fetcher = fetcher + + leaderDone := make(chan error, 1) + go func() { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + drain(res) + leaderDone <- err + }() + + time.Sleep(200 * time.Millisecond) // let the leader take the key + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + start := time.Now() + _, err := proxy.GetOrFetchArtifactFromURL(ctx, "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + blocked := time.Since(start) + + if !errors.Is(err, context.Canceled) { + t.Errorf("waiter error = %v, want context.Canceled", err) + } + if blocked > leaderFetch/4 { + t.Errorf("canceled waiter blocked %v, want well under %v: it is pinned to the shared fetch", + blocked, leaderFetch/4) + } + + // A waiter leaving must not disturb the fetch the others share. + if err := <-leaderDone; err != nil { + t.Fatalf("leader failed after a waiter canceled: %v", err) + } + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + if err != nil { + t.Fatalf("follow-up after leader completed: %v", err) + } + defer func() { _ = res.Reader.Close() }() + if !res.Cached { + t.Error("leader's fetch should have been committed to the cache") + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1", got) + } +} + +// inFlightLen reports how many coalesced fetches are currently registered. +func inFlightLen(p *Proxy) int { + p.fetchMu.Lock() + defer p.fetchMu.Unlock() + return len(p.inFlight) +} + +// TestCoalesce_KeyIsReleasedAfterFetch guards the bug this hand-rolled map can +// have that singleflight could not: a key left behind means later callers join +// a finished entry, see its closed done channel, and are served that stale +// result forever, while the map grows without bound. +func TestCoalesce_KeyIsReleasedAfterFetch(t *testing.T) { + const url = "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz" + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: "artifact bytes", delay: fetchHoldTime} + proxy.Fetcher = fetcher + + _ = runConcurrent(8, func(int) error { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + drain(res) + return err + }) + if n := inFlightLen(proxy); n != 0 { + t.Errorf("in-flight entries after a successful fetch = %d, want 0", n) + } + + // A fresh miss for the same key must start a new fetch, not rejoin the old + // entry. Clearing the cache record forces the miss path again. + if err := proxy.ClearCachedArtifact(context.Background(), "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz"); err != nil { + t.Fatalf("clear cached artifact: %v", err) + } + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + if err != nil { + t.Fatalf("second miss for the same key: %v", err) + } + drain(res) + if got := fetcher.calls.Load(); got != 2 { + t.Errorf("upstream fetches = %d, want 2: the second miss must not reuse the finished entry", got) + } + if n := inFlightLen(proxy); n != 0 { + t.Errorf("in-flight entries at end = %d, want 0", n) + } +} + +// panickingFetcher blows up mid-fetch, after waiters have had time to join. +type panickingFetcher struct{ countingFetcher } + +func (f *panickingFetcher) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) { + return f.FetchWithHeaders(ctx, url, nil) +} + +func (f *panickingFetcher) FetchWithHeaders(context.Context, string, http.Header) (*fetch.Artifact, error) { + f.calls.Add(1) + time.Sleep(fetchHoldTime) + panic("upstream fetch exploded") +} + +// TestCoalesce_PanicInSharedFetchDoesNotStrandWaiters checks the failure mode +// that matters most: a waiter must never be left blocked forever on a fetch +// that died. +func TestCoalesce_PanicInSharedFetchDoesNotStrandWaiters(t *testing.T) { + const url = "https://registry.npmjs.org/pkg/-/pkg-1.0.0.tgz" + proxy, _, _, _ := setupTestProxy(t) + proxy.Fetcher = &panickingFetcher{} + + leaderPanicked := make(chan struct{}) + go func() { + defer func() { + _ = recover() // the panic surfaces in the leader, as it would in a handler + close(leaderPanicked) + }() + res, _ := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + drain(res) + }() + + time.Sleep(fetchHoldTime / 2) // join while the doomed fetch is still running + done := make(chan error, 1) + go func() { + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "pkg", "1.0.0", "pkg-1.0.0.tgz", url) + drain(res) + done <- err + }() + + select { + case err := <-done: + if !errors.Is(err, errSharedFetchAbandoned) { + t.Errorf("waiter error = %v, want errSharedFetchAbandoned", err) + } + case <-time.After(5 * time.Second): + t.Fatal("waiter stranded: a panicking shared fetch never released its waiters") + } + <-leaderPanicked + if n := inFlightLen(proxy); n != 0 { + t.Errorf("in-flight entries after a panic = %d, want 0", n) + } +} diff --git a/internal/handler/coalesce_test.go b/internal/handler/coalesce_test.go new file mode 100644 index 0000000..e95c58f --- /dev/null +++ b/internal/handler/coalesce_test.go @@ -0,0 +1,176 @@ +package handler + +import ( + "bytes" + "context" + "io" + "log/slog" + "net/http" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/storage" + "github.com/git-pkgs/registries/fetch" +) + +// fetchHoldTime holds each stub fetch open long enough that concurrent callers +// reliably overlap inside it. The exact value is not significant. +const fetchHoldTime = 50 * time.Millisecond + +// countingFetcher counts upstream fetches and holds each one open. +type countingFetcher struct { + calls atomic.Int64 + content string + delay time.Duration +} + +func (f *countingFetcher) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) { + return f.FetchWithHeaders(ctx, url, nil) +} + +func (f *countingFetcher) FetchWithHeaders(_ context.Context, _ string, _ http.Header) (*fetch.Artifact, error) { + f.calls.Add(1) + time.Sleep(f.delay) + return &fetch.Artifact{ + Body: io.NopCloser(strings.NewReader(f.content)), + ContentType: "application/gzip", + }, nil +} + +func (f *countingFetcher) Head(context.Context, string) (int64, string, error) { + return 0, "", nil +} + +// TestGetOrFetchArtifactFromURL_ConcurrentMissesCoalesce asserts that N +// simultaneous misses for one artifact produce a single upstream fetch. That is +// the CI shape: parallel jobs installing overlapping dependencies cold. +func TestGetOrFetchArtifactFromURL_ConcurrentMissesCoalesce(t *testing.T) { + const goroutines = 8 + const content = "left-pad tarball bytes" + + proxy, _, _, _ := setupTestProxy(t) + fetcher := &countingFetcher{content: content, delay: fetchHoldTime} + proxy.Fetcher = fetcher + + start := make(chan struct{}) + var wg sync.WaitGroup + errs := make([]error, goroutines) + bodies := make([]string, goroutines) + + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "npm", "left-pad", "1.3.0", "left-pad-1.3.0.tgz", + "https://registry.npmjs.org/left-pad/-/left-pad-1.3.0.tgz") + if err != nil { + errs[i] = err + return + } + defer func() { _ = res.Reader.Close() }() + b, err := io.ReadAll(res.Reader) + errs[i] = err + bodies[i] = string(b) + }(i) + } + + close(start) + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Errorf("goroutine %d: unexpected error: %v", i, err) + } + } + // Every caller must get its own intact copy of the bytes. + for i, b := range bodies { + if b != content { + t.Errorf("goroutine %d: body = %q, want %q", i, b, content) + } + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1 (%d concurrent callers stampeded the upstream)", got, goroutines) + } +} + +// TestGetOrFetchArtifactFromURL_ConcurrentMissesFileStorage runs the same +// scenario against the real file:// backend, the default in production. +// +// Uncoalesced this fails outright, not merely wastefully. Every caller stores +// to one key, and fileblob rewrites a ".attrs" sidecar per key with os.Create, +// truncating in place outside the rename that protects the blob. Decoding that +// sidecar mid-truncate gives "opening reader: EOF", served as a 502. +// +// Only the fetcher is stubbed, because the real one refuses loopback so an +// httptest upstream is unreachable. The storage, where this fails, is real. +func TestGetOrFetchArtifactFromURL_ConcurrentMissesFileStorage(t *testing.T) { + const goroutines = 16 + content := bytes.Repeat([]byte("tarball-bytes-"), 512) + + ctx := context.Background() + dir := t.TempDir() + + db, err := database.Create(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatalf("create database: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + store, err := storage.OpenBucket(ctx, "file://"+filepath.Join(dir, "cache")) + if err != nil { + t.Fatalf("open storage: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + fetcher := &countingFetcher{content: string(content), delay: fetchHoldTime} + proxy := NewProxy(db, store, fetcher, fetch.NewResolver(), + slog.New(slog.NewTextHandler(io.Discard, nil))) + + start := make(chan struct{}) + var wg sync.WaitGroup + errs := make([]error, goroutines) + bodies := make([][]byte, goroutines) + + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + res, err := proxy.GetOrFetchArtifactFromURL(ctx, + "npm", "left-pad", "1.3.0", "left-pad-1.3.0.tgz", + "https://registry.npmjs.org/left-pad/-/left-pad-1.3.0.tgz") + if err != nil { + errs[i] = err + return + } + defer func() { _ = res.Reader.Close() }() + body, readErr := io.ReadAll(res.Reader) + errs[i] = readErr + bodies[i] = body + }(i) + } + + close(start) + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Errorf("caller %d failed: %v", i, err) + } + } + for i, body := range bodies { + if !bytes.Equal(body, content) { + t.Errorf("caller %d got %d bytes, want %d", i, len(body), len(content)) + } + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1", got) + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 856e673..105ad76 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -184,6 +184,11 @@ type Proxy struct { // ScanFetchBaseURL is the address scanners use to reach this proxy to // pull staged artifacts. ScanFetchBaseURL string + + // inFlight coalesces concurrent cache misses for one artifact, so a single + // upstream fetch serves every waiting caller. Keyed by artifactCoalesceKey. + fetchMu sync.Mutex + inFlight map[string]*inflightFetch } // NewProxy creates a new Proxy with the given dependencies. @@ -225,11 +230,10 @@ func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version } metrics.RecordCacheMiss(ecosystem) - stored, storagePath, err := p.fetchAndCache(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL) - if err != nil { - return nil, err - } - return p.openStoredArtifact(ctx, stored, storagePath) + key := artifactCoalesceKey(versionPURL, filename, "", "") + return p.coalesceFetch(ctx, key, func(fetchCtx context.Context) (artifacts.Artifact, string, error) { + return p.fetchAndCache(fetchCtx, ecosystem, name, version, filename, pkgPURL, versionPURL) + }) } // GetCachedArtifact retrieves an artifact from cache without contacting an upstream. @@ -462,8 +466,9 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil } // openStoredArtifact gives one caller its own reader over just-committed -// bytes. A handle cannot be shared: it has one read position, so callers would -// consume each other's bytes and the first Close would break the rest. +// bytes. Callers sharing a fetch cannot share a handle: it has one read +// position, so they would consume each other's bytes and the first Close would +// break the rest. func (p *Proxy) openStoredArtifact(ctx context.Context, artifact artifacts.Artifact, storagePath string) (*CacheResult, error) { readStart := time.Now() reader, err := p.Storage.Open(ctx, storagePath) @@ -481,6 +486,99 @@ func (p *Proxy) openStoredArtifact(ctx context.Context, artifact artifacts.Artif }, nil } +// artifactCoalesceKey identifies one artifact fetch. downloadURL and +// upstreamHash are included so callers expecting different bytes (multiple +// upstreams, or a re-published version) never share a fetch. +func artifactCoalesceKey(versionPURL, filename, downloadURL, upstreamHash string) string { + return strings.Join([]string{versionPURL, filename, downloadURL, upstreamHash}, "\x00") +} + +// errSharedFetchAbandoned is what waiters see if the caller running a shared +// fetch panicked out of it. +var errSharedFetchAbandoned = errors.New("shared upstream fetch did not complete") + +// inflightFetch is one upstream fetch that concurrent callers share. val and +// err are written before done closes and read only after, so the close is the +// handoff. +type inflightFetch struct { + done chan struct{} + val fetchedArtifact + err error +} + +// coalesceFetch runs commit at most once for concurrent callers sharing key, +// then gives each its own reader over the stored bytes. +// +// The first caller in runs the fetch and the rest wait on it. Roles are +// decided under fetchMu rather than inferred afterwards, because the two need +// different cancellation behaviour: a waiter may leave when its own client goes +// away, while the caller running the fetch must see it through so +// storeArtifact's scan-on-disconnect handling still decides the outcome. +// +// commit runs on that caller's context, so cancellation behaves as it did +// uncoalesced and mirroring still relies on it aborting the fetch. If that +// caller goes away, everyone sharing the fetch gets its error and the key is +// released for a later retry. +// +// Every sharing caller still records a cache miss, so the gap between +// proxy_cache_misses_total and upstream fetch observations is what coalescing +// saved. +func (p *Proxy) coalesceFetch(ctx context.Context, key string, commit func(context.Context) (artifacts.Artifact, string, error)) (*CacheResult, error) { + p.fetchMu.Lock() + if p.inFlight == nil { + p.inFlight = make(map[string]*inflightFetch) + } + f, joined := p.inFlight[key] + if !joined { + f = &inflightFetch{done: make(chan struct{})} + p.inFlight[key] = f + } + p.fetchMu.Unlock() + + if !joined { + return p.runSharedFetch(ctx, key, f, commit) + } + + select { + case <-ctx.Done(): + // This caller gave up; the fetch continues for everyone else. + return nil, ctx.Err() + case <-f.done: + } + if f.err != nil { + return nil, f.err + } + return p.openStoredArtifact(ctx, f.val.artifact, f.val.storagePath) +} + +// runSharedFetch performs the fetch that joined callers are waiting on. It is +// never abandoned early, and always releases the key and wakes the waiters. +func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch, commit func(context.Context) (artifacts.Artifact, string, error)) (*CacheResult, error) { + // Set before running so a panicking commit leaves waiters with an error + // rather than a zero-valued artifact. + f.err = errSharedFetchAbandoned + defer func() { + p.fetchMu.Lock() + delete(p.inFlight, key) + p.fetchMu.Unlock() + close(f.done) + }() + + stored, path, err := commit(ctx) + f.val, f.err = fetchedArtifact{artifact: stored, storagePath: path}, err + if err != nil { + return nil, err + } + return p.openStoredArtifact(ctx, stored, path) +} + +// fetchedArtifact is what a shared fetch hands its callers: metadata and a +// storage path, neither holding reader state. +type fetchedArtifact struct { + artifact artifacts.Artifact + storagePath string +} + // runScan generates a signed fetch URL for the just-staged artifact and // asks the configured scanners for a verdict. Returns a wrapped // ErrArtifactBlocked if any scanner blocks, or a scan-infrastructure error. @@ -1114,11 +1212,10 @@ func (p *Proxy) getOrFetchArtifactFromURLWithCachePURLs(ctx context.Context, eco } metrics.RecordCacheMiss(ecosystem) - stored, storagePath, err := p.fetchAndCacheFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash) - if err != nil { - return nil, err - } - return p.openStoredArtifact(ctx, stored, storagePath) + key := artifactCoalesceKey(versionPURL, filename, downloadURL, upstreamHash) + return p.coalesceFetch(ctx, key, func(fetchCtx context.Context) (artifacts.Artifact, string, error) { + return p.fetchAndCacheFromURL(fetchCtx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash) + }) } // getCachedArtifactWithUpstreamHash returns a cached artifact whose recorded diff --git a/internal/handler/handler_test.go b/internal/handler/handler_test.go index 076b74b..b0dbc49 100644 --- a/internal/handler/handler_test.go +++ b/internal/handler/handler_test.go @@ -11,6 +11,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" @@ -29,6 +30,7 @@ import ( // mockStorage implements storage.Storage for testing. type mockStorage struct { + mu sync.Mutex files map[string][]byte storeErr error openErr error @@ -41,6 +43,8 @@ func newMockStorage() *mockStorage { } func (s *mockStorage) Store(_ context.Context, path string, r io.Reader) (int64, string, error) { + s.mu.Lock() + defer s.mu.Unlock() if s.storeErr != nil { return 0, "", s.storeErr } @@ -53,6 +57,8 @@ func (s *mockStorage) Store(_ context.Context, path string, r io.Reader) (int64, } func (s *mockStorage) Open(_ context.Context, path string) (io.ReadCloser, error) { + s.mu.Lock() + defer s.mu.Unlock() if s.openErr != nil { return nil, s.openErr } @@ -64,6 +70,8 @@ func (s *mockStorage) Open(_ context.Context, path string) (io.ReadCloser, error } func (s *mockStorage) Exists(_ context.Context, path string) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() _, ok := s.files[path] return ok, nil } @@ -75,11 +83,15 @@ func (s *mockStorage) Delete(ctx context.Context, path string) error { if err := ctx.Err(); err != nil { return err } + s.mu.Lock() + defer s.mu.Unlock() delete(s.files, path) return nil } func (s *mockStorage) Size(_ context.Context, path string) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() data, ok := s.files[path] if !ok { return 0, storage.ErrNotFound @@ -88,6 +100,8 @@ func (s *mockStorage) Size(_ context.Context, path string) (int64, error) { } func (s *mockStorage) UsedSpace(_ context.Context) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() var total int64 for _, data := range s.files { total += int64(len(data))