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 a78393d..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,7 +230,10 @@ func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version } metrics.RecordCacheMiss(ecosystem) - return p.fetchAndCache(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL) + 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. @@ -359,14 +367,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 +394,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 +413,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 +425,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 +444,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 +462,14 @@ 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. 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) metrics.RecordStorageOperation("read", time.Since(readStart)) @@ -463,11 +481,104 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil return &CacheResult{ Reader: reader, - Artifact: sharedArtifact, + Artifact: artifact, Cached: false, }, 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. @@ -1101,7 +1212,10 @@ func (p *Proxy) getOrFetchArtifactFromURLWithCachePURLs(ctx context.Context, eco } metrics.RecordCacheMiss(ecosystem) - return p.fetchAndCacheFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash) + 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 @@ -1128,7 +1242,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 +1252,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) 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))