diff --git a/go.mod b/go.mod index 670f884e..dd053fda 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/git-pkgs/integrity v0.2.0 github.com/git-pkgs/magic v0.5.0 github.com/git-pkgs/purl v0.1.21 - github.com/git-pkgs/registries v0.9.3 + github.com/git-pkgs/registries v0.10.0 github.com/git-pkgs/spdx v0.3.3 github.com/git-pkgs/vers v0.7.2 github.com/git-pkgs/vulns v0.2.4 diff --git a/go.sum b/go.sum index d2ed2447..9e5b98d6 100644 --- a/go.sum +++ b/go.sum @@ -143,8 +143,8 @@ github.com/git-pkgs/pom v0.1.8 h1:mXCuWw6XkrKKZb6IurRZyIznYLPhV5AYg5ZO38Uj/l8= github.com/git-pkgs/pom v0.1.8/go.mod h1:sQBeYRzjCUn77fcABPn6I/a1aBHFxm55jTmkNxYtGtw= github.com/git-pkgs/purl v0.1.21 h1:PDZUX43lxIiXPn2eHcsNiiiwQo+HajntZ/4TwSvCGlw= github.com/git-pkgs/purl v0.1.21/go.mod h1:4dOmris48ehjiKmWBqSmMHguUQht48BLeXW+hQUnuj0= -github.com/git-pkgs/registries v0.9.3 h1:nXIP//6kVqLKqYXOWhEMoWUkJPDAZYVLRUYfHz3jM2Y= -github.com/git-pkgs/registries v0.9.3/go.mod h1:BpXlfRCfu4S5bTzm7u7igppVu94mXgll/PENe0AjKmI= +github.com/git-pkgs/registries v0.10.0 h1:0goaC5mxw9MglS/XOm2De7fkPBWCCerelm7RPrPh9hQ= +github.com/git-pkgs/registries v0.10.0/go.mod h1:bKxgdy/PwnzQXL2FnOktmSQHTKk6B85MA4VtvvYItPc= github.com/git-pkgs/spdx v0.3.3 h1:xXlUoZzjDSkwxC6mWHfaW+kc8a0+LH6d+gGHuZWftqI= github.com/git-pkgs/spdx v0.3.3/go.mod h1:n9rAicgw+Wqtfursju6oKaAL6cUCTiIS926mb1KWoKU= github.com/git-pkgs/vers v0.7.2 h1:Ctex1O+Ai/FONmyDaw37EolBXg0og6UdxbCjjYbeRHs= diff --git a/internal/handler/artifact_size.go b/internal/handler/artifact_size.go deleted file mode 100644 index d4c733b3..00000000 --- a/internal/handler/artifact_size.go +++ /dev/null @@ -1,76 +0,0 @@ -package handler - -import ( - "context" - "errors" - "fmt" - "io" - "net/http" - - "github.com/git-pkgs/registries/fetch" -) - -// ErrArtifactTooLarge is returned for an upstream artifact larger than the -// configured artifact size limit. -var ErrArtifactTooLarge = errors.New("artifact exceeds size limit") - -// LimitArtifactSize wraps fetcher so no artifact over maxSize bytes is -// fetched. An artifact whose declared size is over the limit is refused -// before its body is read; one of unknown size fails once reading passes the -// limit, so the store or stream consuming it fails and nothing is cached. -// A maxSize of zero or less returns fetcher unchanged. -func LimitArtifactSize(fetcher fetch.FetcherInterface, maxSize int64) fetch.FetcherInterface { //nolint:ireturn // returns fetcher itself when there is no limit - if maxSize <= 0 { - return fetcher - } - return &sizeLimitedFetcher{FetcherInterface: fetcher, maxSize: maxSize} -} - -type sizeLimitedFetcher struct { - fetch.FetcherInterface - maxSize int64 -} - -func (f *sizeLimitedFetcher) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) { - return f.limit(f.FetcherInterface.Fetch(ctx, url)) -} - -func (f *sizeLimitedFetcher) FetchWithHeaders(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) { - return f.limit(f.FetcherInterface.FetchWithHeaders(ctx, url, headers)) -} - -func (f *sizeLimitedFetcher) limit(artifact *fetch.Artifact, err error) (*fetch.Artifact, error) { - if err != nil { - return nil, err - } - if artifact.Size > f.maxSize { - _ = artifact.Body.Close() - return nil, fmt.Errorf("%w: upstream declared %d bytes, limit is %d", ErrArtifactTooLarge, artifact.Size, f.maxSize) - } - limited := *artifact - limited.Body = &sizeLimitedBody{ReadCloser: artifact.Body, remaining: f.maxSize} - return &limited, nil -} - -// sizeLimitedBody fails with ErrArtifactTooLarge once more than the limit has -// been read. It reads one byte past the limit to tell an artifact of exactly -// the limit from a larger one. -type sizeLimitedBody struct { - io.ReadCloser - remaining int64 -} - -func (b *sizeLimitedBody) Read(p []byte) (int, error) { - if b.remaining < 0 { - return 0, ErrArtifactTooLarge - } - if int64(len(p)) > b.remaining+1 { - p = p[:b.remaining+1] - } - n, err := b.ReadCloser.Read(p) - b.remaining -= int64(n) - if b.remaining < 0 { - return n + int(b.remaining), ErrArtifactTooLarge - } - return n, err -} diff --git a/internal/handler/artifact_size_test.go b/internal/handler/artifact_size_test.go index f6fe9af1..68dc25d5 100644 --- a/internal/handler/artifact_size_test.go +++ b/internal/handler/artifact_size_test.go @@ -6,83 +6,36 @@ import ( "io" "net/http" "net/http/httptest" - "strings" "testing" "github.com/git-pkgs/registries/fetch" ) -type closeRecorder struct { - io.Reader - closed bool -} - -func (c *closeRecorder) Close() error { - c.closed = true - return nil -} - -func TestLimitArtifactSizeUnlimitedReturnsFetcher(t *testing.T) { - fetcher := &mockFetcher{} - if got := LimitArtifactSize(fetcher, 0); got != fetcher { - t.Errorf("LimitArtifactSize(f, 0) = %T, want the fetcher itself", got) - } -} - -func TestLimitArtifactSizeRefusesDeclaredOversize(t *testing.T) { - body := &closeRecorder{Reader: strings.NewReader("0123456789")} - limited := LimitArtifactSize(&mockFetcher{artifact: &fetch.Artifact{Body: body, Size: 10}}, 4) - - _, err := limited.Fetch(context.Background(), "https://registry.example/a.tgz") - if !errors.Is(err, ErrArtifactTooLarge) { - t.Fatalf("Fetch error = %v, want ErrArtifactTooLarge", err) - } - if !body.closed { - t.Error("refused body was not closed") - } -} - -func TestLimitArtifactSizeFailsUndeclaredOversizeWhileReading(t *testing.T) { - limited := LimitArtifactSize(&mockFetcher{artifact: &fetch.Artifact{ - Body: io.NopCloser(strings.NewReader("0123456789")), Size: -1, - }}, 4) - - artifact, err := limited.Fetch(context.Background(), "https://registry.example/a.tgz") - if err != nil { - t.Fatalf("Fetch failed: %v", err) - } - if _, err := io.ReadAll(artifact.Body); !errors.Is(err, ErrArtifactTooLarge) { - t.Errorf("reading body error = %v, want ErrArtifactTooLarge", err) - } -} - -func TestLimitArtifactSizeAllowsArtifactAtLimit(t *testing.T) { - limited := LimitArtifactSize(&mockFetcher{artifact: &fetch.Artifact{ - Body: io.NopCloser(strings.NewReader("0123")), Size: -1, - }}, 4) - - artifact, err := limited.FetchWithHeaders(context.Background(), "https://registry.example/a.tgz", nil) - if err != nil { - t.Fatalf("Fetch failed: %v", err) - } - data, err := io.ReadAll(artifact.Body) - if err != nil || string(data) != "0123" { - t.Errorf("body = %q, %v; want %q, nil", data, err, "0123") - } +func artifactSizeFetcher(t *testing.T) (*fetch.Fetcher, string) { + t.Helper() + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.(http.Flusher).Flush() + _, _ = io.WriteString(w, "0123456789") + })) + t.Cleanup(upstream.Close) + f := fetch.NewFetcher(fetch.WithHTTPClient(upstream.Client()), fetch.WithMaxArtifactSize(4)) + t.Cleanup(func() { _ = f.Close() }) + return f, upstream.URL } func TestOversizedArtifactIsNotCached(t *testing.T) { - proxy, db, store, fetcher := setupTestProxy(t) - fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("0123456789")), Size: -1} - proxy.Fetcher = LimitArtifactSize(fetcher, 4) + proxy, db, store, _ := setupTestProxy(t) + limited, upstream := artifactSizeFetcher(t) + proxy.Fetcher = limited - _, err := proxy.GetOrFetchArtifactFromURL(context.Background(), "npm", "big", "1.0.0", "big-1.0.0.tgz", "https://registry.example/big-1.0.0.tgz") - if !errors.Is(err, ErrArtifactTooLarge) { - t.Fatalf("GetOrFetchArtifactFromURL error = %v, want ErrArtifactTooLarge", err) + _, err := proxy.GetOrFetchArtifactFromURL(context.Background(), "npm", "big", "1.0.0", "big-1.0.0.tgz", upstream+"/big-1.0.0.tgz") + if !errors.Is(err, fetch.ErrArtifactTooLarge) { + t.Fatalf("GetOrFetchArtifactFromURL error = %v, want fetch.ErrArtifactTooLarge", err) } if len(store.files) != 0 { t.Errorf("storage holds %d objects after an oversized fetch, want 0", len(store.files)) } + fetchErr := err record, err := db.GetArtifact("pkg:npm/big@1.0.0", "big-1.0.0.tgz") if err != nil { t.Fatalf("GetArtifact failed: %v", err) @@ -92,24 +45,24 @@ func TestOversizedArtifactIsNotCached(t *testing.T) { } rec := httptest.NewRecorder() - proxy.serveArtifactError(rec, err, "failed to fetch package") + proxy.serveArtifactError(rec, fetchErr, "failed to fetch package") if rec.Code != http.StatusBadGateway { t.Errorf("status = %d, want %d", rec.Code, http.StatusBadGateway) } } func TestOversizedStreamedArtifactFailsTheStream(t *testing.T) { - proxy, _, _, fetcher := setupTestProxy(t) + proxy, _, _, _ := setupTestProxy(t) proxy.StreamArtifacts = true - fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("0123456789")), Size: -1} - proxy.Fetcher = LimitArtifactSize(fetcher, 4) + limited, upstream := artifactSizeFetcher(t) + proxy.Fetcher = limited - result, err := proxy.GetOrFetchArtifactFromURL(context.Background(), "npm", "big", "1.0.0", "big-1.0.0.tgz", "https://registry.example/big-1.0.0.tgz") + result, err := proxy.GetOrFetchArtifactFromURL(context.Background(), "npm", "big", "1.0.0", "big-1.0.0.tgz", upstream+"/big-1.0.0.tgz") if err != nil { t.Fatalf("GetOrFetchArtifactFromURL failed: %v", err) } defer func() { _ = result.Reader.Close() }() - if _, err := io.ReadAll(result.Reader); !errors.Is(err, ErrArtifactTooLarge) { - t.Errorf("streamed body error = %v, want ErrArtifactTooLarge", err) + if _, err := io.ReadAll(result.Reader); !errors.Is(err, fetch.ErrArtifactTooLarge) { + t.Errorf("streamed body error = %v, want fetch.ErrArtifactTooLarge", err) } } diff --git a/internal/handler/download_test.go b/internal/handler/download_test.go index 2ac8ace6..876185e5 100644 --- a/internal/handler/download_test.go +++ b/internal/handler/download_test.go @@ -12,10 +12,10 @@ import ( "time" "github.com/git-pkgs/proxy/internal/database" - "github.com/git-pkgs/proxy/internal/httpclient" "github.com/git-pkgs/proxy/internal/metrics" "github.com/git-pkgs/proxy/internal/storage" "github.com/git-pkgs/purl" + registryclient "github.com/git-pkgs/registries/client" "github.com/git-pkgs/registries/fetch" ) @@ -1278,7 +1278,7 @@ func TestRPMHandler_DownloadCacheMiss(t *testing.T) { } func TestMavenHandler_UpstreamHeldReturns503(t *testing.T) { - held := &httpclient.HeldError{Host: "repo1.maven.org", Until: time.Now().Add(90 * time.Second), RetryAfter: 90 * time.Second} + held := ®istryclient.HeldError{Host: "repo1.maven.org", Until: time.Now().Add(90 * time.Second), RetryAfter: 90 * time.Second} t.Run("artifact", func(t *testing.T) { proxy, _, _, fetcher := setupTestProxy(t) @@ -1326,7 +1326,7 @@ func (fn roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) func TestMavenHandler_HeldArtifactMissesSkipCircuitBreaker(t *testing.T) { proxy, _, _, _ := setupTestProxy(t) upstreamCalls := 0 - gate := httpclient.NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { + gate := registryclient.NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { upstreamCalls++ return &http.Response{ StatusCode: http.StatusTooManyRequests, @@ -1366,7 +1366,7 @@ func TestMavenHandler_HeldArtifactMissesSkipCircuitBreaker(t *testing.T) { } func TestHandlers_UpstreamHeldMetadataReturns503(t *testing.T) { - held := &httpclient.HeldError{Host: "upstream.example", Until: time.Now().Add(42 * time.Second), RetryAfter: 42 * time.Second} + held := ®istryclient.HeldError{Host: "upstream.example", Until: time.Now().Add(42 * time.Second), RetryAfter: 42 * time.Second} tests := []struct { name string handler func(*Proxy) http.Handler diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 62d63aec..dcdebeb6 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -20,12 +20,12 @@ import ( "github.com/git-pkgs/cooldown" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/denylist" - "github.com/git-pkgs/proxy/internal/httpclient" "github.com/git-pkgs/proxy/internal/metrics" "github.com/git-pkgs/proxy/internal/packageurl" "github.com/git-pkgs/proxy/internal/scanner" "github.com/git-pkgs/proxy/internal/storage" "github.com/git-pkgs/purl" + registryclient "github.com/git-pkgs/registries/client" "github.com/git-pkgs/registries/fetch" "github.com/opencontainers/go-digest" ) @@ -178,7 +178,7 @@ type Proxy struct { // RateLimits is the gate under the upstream clients. Artifact fetches // check it before calling Fetcher, so held hosts never reach its breaker. - RateLimits *httpclient.RateLimitGate + RateLimits *registryclient.RateLimitGate // HostedNPMScopes and HostedComposerVendors name namespaces served from // pushed packages alone, without asking the upstream. diff --git a/internal/handler/upstream_held.go b/internal/handler/upstream_held.go index fa2ff968..1097e7ef 100644 --- a/internal/handler/upstream_held.go +++ b/internal/handler/upstream_held.go @@ -6,14 +6,14 @@ import ( "net/http" "strconv" - "github.com/git-pkgs/proxy/internal/httpclient" + registryclient "github.com/git-pkgs/registries/client" ) // writeUpstreamHeld answers a request the rate limit gate refused to send // with 503 and the Retry-After the upstream asked for, and reports whether // err was such a refusal. func writeUpstreamHeld(w http.ResponseWriter, err error) bool { - var held *httpclient.HeldError + var held *registryclient.HeldError if !errors.As(err, &held) { return false } diff --git a/internal/httpclient/ratelimit.go b/internal/httpclient/ratelimit.go deleted file mode 100644 index 53e3e8ef..00000000 --- a/internal/httpclient/ratelimit.go +++ /dev/null @@ -1,228 +0,0 @@ -package httpclient - -import ( - "errors" - "fmt" - "log/slog" - "net/http" - "net/url" - "strconv" - "strings" - "sync" - "time" -) - -const ( - // holdWithoutHeader is the first hold for a 429 that carries no usable - // Retry-After; each further such 429 after a hold expires doubles it. - holdWithoutHeader = time.Minute - maxHoldWithoutHeader = 15 * time.Minute //nolint:mnd // doubling stops here - // maxHold caps what an upstream's Retry-After can ask for. - maxHold = time.Hour -) - -// ErrUpstreamHeld marks a request the gate did not send because the upstream -// is rate limiting this proxy. Match it with errors.Is; the *HeldError in -// the chain carries the host and how long to wait. -var ErrUpstreamHeld = errors.New("upstream is rate limiting this proxy") - -// HeldError is returned in place of a response while a host is held. -type HeldError struct { - Host string - Until time.Time - RetryAfter time.Duration -} - -func (e *HeldError) Error() string { - return fmt.Sprintf("%s is rate limiting this proxy, holding requests for %s", e.Host, e.RetryAfter.Round(time.Second)) -} - -func (e *HeldError) Is(target error) bool { return target == ErrUpstreamHeld } - -type hostHold struct { - until time.Time - strikes int -} - -// RateLimitGate stops sending to a host that answered 429, or 503 with a -// Retry-After, for as long as it asked, and fails those requests locally -// with a *HeldError instead. Clients and fetchers above it see an error -// rather than a slow upstream, so they fall back to cache or fail fast -// without adding to the traffic the upstream is already refusing. The first -// non-limited response after the hold ends releases it. -type RateLimitGate struct { - base http.RoundTripper - Logger *slog.Logger - // OnHold is called with the host each time a hold starts. - OnHold func(host string) - now func() time.Time - - mu sync.Mutex - holds map[string]*hostHold -} - -// NewRateLimitGate wraps base. A nil base uses http.DefaultTransport. -func NewRateLimitGate(base http.RoundTripper) *RateLimitGate { - if base == nil { - base = http.DefaultTransport - } - return &RateLimitGate{ - base: base, - now: time.Now, - holds: make(map[string]*hostHold), - } -} - -// RoundTrip implements http.RoundTripper. -func (g *RateLimitGate) RoundTrip(req *http.Request) (*http.Response, error) { - host := hostKey(req.URL) - if until, held := g.heldUntil(host); held { - return nil, &HeldError{Host: host, Until: until, RetryAfter: until.Sub(g.now())} - } - - resp, err := g.base.RoundTrip(req) - if err != nil { - return nil, err - } - - retryAfter := resp.Header.Get("Retry-After") - switch { - case resp.StatusCode == http.StatusTooManyRequests, - resp.StatusCode == http.StatusServiceUnavailable && retryAfter != "": - g.hold(host, retryAfter) - default: - g.release(host) - } - return resp, nil -} - -// Held returns a *HeldError when requests to rawURL's host are being held, -// and nil otherwise. It is safe to call on a nil gate. -func (g *RateLimitGate) Held(rawURL string) error { - if g == nil { - return nil - } - u, err := url.Parse(rawURL) - if err != nil { - return nil - } - host := hostKey(u) - until, held := g.heldUntil(host) - if !held { - return nil - } - return &HeldError{Host: host, Until: until, RetryAfter: until.Sub(g.now())} -} - -// Holds returns the hosts currently held and when each hold ends. It is -// safe to call on a nil gate. -func (g *RateLimitGate) Holds() map[string]time.Time { - if g == nil { - return nil - } - now := g.now() - g.mu.Lock() - defer g.mu.Unlock() - held := make(map[string]time.Time) - for host, h := range g.holds { - if now.Before(h.until) { - held[host] = h.until - } - } - return held -} - -func (g *RateLimitGate) heldUntil(host string) (time.Time, bool) { - now := g.now() - g.mu.Lock() - defer g.mu.Unlock() - h, ok := g.holds[host] - if !ok || !now.Before(h.until) { - return time.Time{}, false - } - return h.until, true -} - -func (g *RateLimitGate) hold(host, header string) { - now := g.now() - - g.mu.Lock() - h, ok := g.holds[host] - if !ok { - h = &hostHold{} - g.holds[host] = h - } - // Responses already in flight when a hold starts all report the same - // rate limiting; only a 429 after the hold ran out is a further strike. - if !now.Before(h.until) { - h.strikes++ - } - d, fromHeader := parseRetryAfter(header, now) - if !fromHeader { - d = min(holdWithoutHeader<<(h.strikes-1), maxHoldWithoutHeader) - } - d = min(d, maxHold) - until := now.Add(d) - if until.After(h.until) { - h.until = until - } - g.mu.Unlock() - - if g.Logger != nil { - source := "default" - if fromHeader { - source = "retry-after" - } - g.Logger.Warn("upstream rate limited this proxy, holding its requests", - "host", host, "hold", d.Round(time.Second).String(), "source", source, - "until", until.UTC().Format(time.RFC3339)) - } - if g.OnHold != nil { - g.OnHold(host) - } -} - -// release drops host's hold once it has ended. Nothing is sent during a -// hold, so a success arriving while one is active was sent before it began. -func (g *RateLimitGate) release(host string) { - now := g.now() - g.mu.Lock() - h, had := g.holds[host] - if had && now.Before(h.until) { - g.mu.Unlock() - return - } - delete(g.holds, host) - g.mu.Unlock() - if had && g.Logger != nil { - g.Logger.Info("upstream rate limit cleared", "host", host) - } -} - -// parseRetryAfter reads a Retry-After value as delay seconds or an HTTP -// date. It reports false for anything unusable, including a zero or past -// value, so the caller applies its default instead. -func parseRetryAfter(value string, now time.Time) (time.Duration, bool) { - value = strings.TrimSpace(value) - if value == "" { - return 0, false - } - if secs, err := strconv.ParseInt(value, 10, 64); err == nil { - if secs <= 0 { - return 0, false - } - return time.Duration(secs) * time.Second, true - } - at, err := http.ParseTime(value) - if err != nil || !at.After(now) { - return 0, false - } - return at.Sub(now), true -} - -func hostKey(u *url.URL) string { - if u == nil { - return "" - } - return strings.ToLower(u.Host) -} diff --git a/internal/httpclient/ratelimit_test.go b/internal/httpclient/ratelimit_test.go deleted file mode 100644 index 8ad7e9c4..00000000 --- a/internal/httpclient/ratelimit_test.go +++ /dev/null @@ -1,257 +0,0 @@ -package httpclient - -import ( - "errors" - "io" - "net/http" - "strings" - "testing" - "time" -) - -type gateHarness struct { - gate *RateLimitGate - now time.Time - calls map[string]int - // respond returns the status and Retry-After the fake upstream answers with. - respond func(req *http.Request) (int, string) -} - -func newGateHarness() *gateHarness { - h := &gateHarness{ - now: time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC), - calls: map[string]int{}, - respond: func(*http.Request) (int, string) { - return http.StatusOK, "" - }, - } - h.gate = NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { - h.calls[req.URL.Host]++ - status, retryAfter := h.respond(req) - resp := &http.Response{ - StatusCode: status, - Header: http.Header{}, - Body: io.NopCloser(strings.NewReader("")), - Request: req, - } - if retryAfter != "" { - resp.Header.Set("Retry-After", retryAfter) - } - return resp, nil - })) - h.gate.now = func() time.Time { return h.now } - return h -} - -func (h *gateHarness) get(t *testing.T, rawURL string) (*http.Response, error) { - t.Helper() - req, err := http.NewRequest(http.MethodGet, rawURL, nil) - if err != nil { - t.Fatalf("building request: %v", err) - } - resp, err := h.gate.RoundTrip(req) - if resp != nil { - _ = resp.Body.Close() - } - return resp, err -} - -func TestRateLimitGate_HoldsHostForRetryAfter(t *testing.T) { - h := newGateHarness() - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "785" } - - resp, err := h.get(t, "https://repo1.maven.org/maven2/a/b/maven-metadata.xml") - if err != nil || resp.StatusCode != http.StatusTooManyRequests { - t.Fatalf("first request: resp=%v err=%v, want the upstream 429 passed through", resp, err) - } - - h.respond = func(*http.Request) (int, string) { return http.StatusOK, "" } - _, err = h.get(t, "https://repo1.maven.org/maven2/a/b/b-1.0.jar") - var held *HeldError - if !errors.As(err, &held) || !errors.Is(err, ErrUpstreamHeld) { - t.Fatalf("second request error = %v, want *HeldError", err) - } - if held.Host != "repo1.maven.org" || held.RetryAfter != 785*time.Second { - t.Errorf("held = %+v, want repo1.maven.org for 785s", held) - } - if h.calls["repo1.maven.org"] != 1 { - t.Errorf("upstream calls = %d, want 1: held requests must not reach it", h.calls["repo1.maven.org"]) - } - - if _, err := h.get(t, "https://plugins.gradle.org/m2/x.pom"); err != nil { - t.Errorf("other host: %v, want unaffected", err) - } - - holds := h.gate.Holds() - if until, ok := holds["repo1.maven.org"]; !ok || !until.Equal(h.now.Add(785*time.Second)) { - t.Errorf("Holds() = %v, want repo1.maven.org until now+785s", holds) - } -} - -func TestRateLimitGate_ReleasesWhenHoldExpiresAndUpstreamServes(t *testing.T) { - h := newGateHarness() - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "60" } - _, _ = h.get(t, "https://repo1.maven.org/x") - - h.now = h.now.Add(59 * time.Second) - if _, err := h.get(t, "https://repo1.maven.org/x"); !errors.Is(err, ErrUpstreamHeld) { - t.Fatalf("at 59s: err = %v, want still held", err) - } - - h.now = h.now.Add(time.Second) - h.respond = func(*http.Request) (int, string) { return http.StatusOK, "" } - resp, err := h.get(t, "https://repo1.maven.org/x") - if err != nil || resp.StatusCode != http.StatusOK { - t.Fatalf("at 60s: resp=%v err=%v, want the request sent", resp, err) - } - if len(h.gate.Holds()) != 0 { - t.Errorf("Holds() = %v after a success, want empty", h.gate.Holds()) - } -} - -func TestRateLimitGate_LateSuccessKeepsHold(t *testing.T) { - h := newGateHarness() - // The slow request is in flight when the limited one starts the hold, - // and its 200 arrives afterwards. - h.respond = func(req *http.Request) (int, string) { - if req.URL.Path == "/limited" { - return http.StatusTooManyRequests, "120" - } - _, _ = h.get(t, "https://repo1.maven.org/limited") - return http.StatusOK, "" - } - - resp, err := h.get(t, "https://repo1.maven.org/slow") - if err != nil || resp.StatusCode != http.StatusOK { - t.Fatalf("slow request: resp=%v err=%v, want its 200", resp, err) - } - if until, ok := h.gate.Holds()["repo1.maven.org"]; !ok || !until.Equal(h.now.Add(120*time.Second)) { - t.Fatalf("Holds() = %v after a late 200, want repo1.maven.org until now+120s", h.gate.Holds()) - } - if _, err := h.get(t, "https://repo1.maven.org/x"); !errors.Is(err, ErrUpstreamHeld) { - t.Errorf("next request err = %v, want held", err) - } -} - -func TestRateLimitGate_Held(t *testing.T) { - var nilGate *RateLimitGate - if err := nilGate.Held("https://repo1.maven.org/x"); err != nil { - t.Errorf("nil gate Held() = %v, want nil", err) - } - - h := newGateHarness() - if err := h.gate.Held("https://repo1.maven.org/x"); err != nil { - t.Errorf("before any limit: Held() = %v, want nil", err) - } - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "45" } - _, _ = h.get(t, "https://repo1.maven.org/x") - - var held *HeldError - if err := h.gate.Held("https://REPO1.maven.org/a/b.jar"); !errors.As(err, &held) || held.RetryAfter != 45*time.Second { - t.Errorf("Held() = %v, want *HeldError for 45s", err) - } - if err := h.gate.Held("https://plugins.gradle.org/m2/x.pom"); err != nil { - t.Errorf("other host: Held() = %v, want nil", err) - } - h.now = h.now.Add(45 * time.Second) - if err := h.gate.Held("https://repo1.maven.org/x"); err != nil { - t.Errorf("after the hold: Held() = %v, want nil", err) - } -} - -func TestRateLimitGate_DefaultHoldDoublesWithoutHeader(t *testing.T) { - h := newGateHarness() - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "" } - - wantHolds := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute} - for i, want := range wantHolds { - _, _ = h.get(t, "https://repo1.maven.org/x") - until := h.gate.Holds()["repo1.maven.org"] - if got := until.Sub(h.now); got != want { - t.Errorf("strike %d: hold = %s, want %s", i+1, got, want) - } - h.now = until - } -} - -func TestRateLimitGate_ConcurrentLimitedResponsesAreOneStrike(t *testing.T) { - h := newGateHarness() - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "" } - - // Two requests already in flight when the limit starts both see 429. - // The gate only consults the hold before sending, so drive the base - // directly for the second. - _, _ = h.get(t, "https://repo1.maven.org/x") - req, _ := http.NewRequest(http.MethodGet, "https://repo1.maven.org/y", nil) - h.gate.hold(hostKey(req.URL), "") - - if got := h.gate.Holds()["repo1.maven.org"].Sub(h.now); got != time.Minute { - t.Errorf("hold = %s after two concurrent 429s, want the single-strike %s", got, time.Minute) - } -} - -func TestRateLimitGate_CapsUpstreamRetryAfter(t *testing.T) { - h := newGateHarness() - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "86400" } - _, _ = h.get(t, "https://repo1.maven.org/x") - if got := h.gate.Holds()["repo1.maven.org"].Sub(h.now); got != maxHold { - t.Errorf("hold = %s, want capped at %s", got, maxHold) - } -} - -func TestRateLimitGate_ServiceUnavailableNeedsRetryAfter(t *testing.T) { - h := newGateHarness() - h.respond = func(*http.Request) (int, string) { return http.StatusServiceUnavailable, "" } - _, _ = h.get(t, "https://repo1.maven.org/x") - if len(h.gate.Holds()) != 0 { - t.Errorf("503 without Retry-After created hold %v, want none", h.gate.Holds()) - } - - h.respond = func(*http.Request) (int, string) { return http.StatusServiceUnavailable, "30" } - _, _ = h.get(t, "https://repo1.maven.org/x") - if got := h.gate.Holds()["repo1.maven.org"].Sub(h.now); got != 30*time.Second { - t.Errorf("hold = %s, want 30s from the 503's Retry-After", got) - } -} - -func TestRateLimitGate_ReportsHoldsAndNilSafe(t *testing.T) { - var nilGate *RateLimitGate - if holds := nilGate.Holds(); holds != nil { - t.Errorf("nil gate Holds() = %v, want nil", holds) - } - - h := newGateHarness() - var reported []string - h.gate.OnHold = func(host string) { - reported = append(reported, host) - } - h.respond = func(*http.Request) (int, string) { return http.StatusTooManyRequests, "10" } - _, _ = h.get(t, "https://repo1.maven.org/x") - if len(reported) != 1 || reported[0] != "repo1.maven.org" { - t.Errorf("OnHold reported %v, want [repo1.maven.org]", reported) - } -} - -func TestParseRetryAfter(t *testing.T) { - now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) - tests := []struct { - value string - want time.Duration - ok bool - }{ - {"785", 785 * time.Second, true}, - {" 30 ", 30 * time.Second, true}, - {"0", 0, false}, - {"-5", 0, false}, - {"", 0, false}, - {"soon", 0, false}, - {now.Add(90 * time.Second).Format(http.TimeFormat), 90 * time.Second, true}, - {now.Add(-time.Minute).Format(http.TimeFormat), 0, false}, - } - for _, tt := range tests { - got, ok := parseRetryAfter(tt.value, now) - if got != tt.want || ok != tt.ok { - t.Errorf("parseRetryAfter(%q) = (%s, %v), want (%s, %v)", tt.value, got, ok, tt.want, tt.ok) - } - } -} diff --git a/internal/server/breakers_test.go b/internal/server/breakers_test.go index bb5166ee..3a9e7640 100644 --- a/internal/server/breakers_test.go +++ b/internal/server/breakers_test.go @@ -12,8 +12,8 @@ import ( "testing" "time" - upstreamhttp "github.com/git-pkgs/proxy/internal/httpclient" "github.com/git-pkgs/proxy/internal/metrics" + registryclient "github.com/git-pkgs/registries/client" "github.com/git-pkgs/registries/fetch" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" @@ -352,7 +352,7 @@ func TestHealthEndpoint_ReportsUpstreamHolds(t *testing.T) { ts := newTestServer(t) defer ts.close() - gate := upstreamhttp.NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { + gate := registryclient.NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { return &http.Response{ StatusCode: http.StatusTooManyRequests, Header: http.Header{"Retry-After": []string{"600"}}, diff --git a/internal/server/registry_policies_test.go b/internal/server/registry_policies_test.go new file mode 100644 index 00000000..881356a3 --- /dev/null +++ b/internal/server/registry_policies_test.go @@ -0,0 +1,79 @@ +package server + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/config" + registryclient "github.com/git-pkgs/registries/client" + "github.com/git-pkgs/registries/fetch" +) + +func TestMirrorProxySharesHoldAfterBearerRetry(t *testing.T) { + var calls atomic.Int32 + var upstream *httptest.Server + upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.URL.Path == "/token" { + _, _ = io.WriteString(w, `{"token":"fixture-token"}`) + return + } + if r.Header.Get("Authorization") != "Bearer fixture-token" { + w.Header().Set("WWW-Authenticate", fmt.Sprintf(`Bearer realm="%s/token",service="test",scope="repository:test:pull"`, upstream.URL)) + w.WriteHeader(http.StatusUnauthorized) + return + } + w.Header().Set("Retry-After", "9223372036854775807") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer upstream.Close() + cfg := config.Default() + cfg.Upstream.AllowLoopback = true + proxy, _ := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + response, err := proxy.HTTPClient.Get(upstream.URL + "/v2/test/manifests/latest") + if err != nil { + t.Fatal(err) + } + _ = response.Body.Close() + if response.StatusCode != http.StatusTooManyRequests || calls.Load() != 3 { + t.Fatalf("bearer retry: status %d, calls %d", response.StatusCode, calls.Load()) + } + _, err = proxy.Fetcher.Fetch(context.Background(), upstream.URL+"/v2/test/blobs/content") + if !errors.Is(err, registryclient.ErrUpstreamHeld) { + t.Fatalf("artifact fetch = %v", err) + } + metadata := registryclient.NewClient(registryclient.WithHTTPClient(proxy.HTTPClient)) + metadata.BaseDelay = time.Hour + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err = metadata.GetBody(ctx, upstream.URL+"/metadata") + if !errors.Is(err, registryclient.ErrUpstreamHeld) || calls.Load() != 3 { + t.Fatalf("metadata fetch = %v, upstream calls %d", err, calls.Load()) + } + var held *registryclient.HeldError + if !errors.As(proxy.RateLimits.Held(upstream.URL), &held) || held.RetryAfter < 59*time.Minute || len(proxy.RateLimits.Holds()) != 1 { + t.Fatalf("shared hold = %+v", held) + } +} + +func TestMirrorProxyAppliesArtifactSizeOption(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "12345") + })) + defer upstream.Close() + cfg := config.Default() + cfg.Upstream.AllowLoopback = true + cfg.ArtifactMaxSize = "4" + proxy, _ := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + if _, err := proxy.Fetcher.Fetch(context.Background(), upstream.URL+"/artifact"); !errors.Is(err, fetch.ErrArtifactTooLarge) { + t.Fatalf("Fetch = %v", err) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index d9699cd4..5eaaa25d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -81,6 +81,7 @@ import ( "github.com/git-pkgs/proxy/internal/packageurl" "github.com/git-pkgs/proxy/internal/scanner" "github.com/git-pkgs/proxy/internal/storage" + registryclient "github.com/git-pkgs/registries/client" "github.com/git-pkgs/registries/fetch" "github.com/git-pkgs/registries/safehttp" "github.com/git-pkgs/spdx" @@ -117,7 +118,7 @@ type Server struct { accessLog *accesslog.Logger ecr *ecrTokens breakers *breakerMonitor - rateLimits *upstreamhttp.RateLimitGate + rateLimits *registryclient.RateLimitGate ecoStats ecosystemStatsCache sources sourceTracker @@ -227,7 +228,7 @@ func (s *Server) serve(listener net.Listener) error { metadataClient, artifactClient := s.newUpstreamClients(serverWriteTimeout) // Create shared components with circuit breaker. - baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(artifactClient)) + baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(artifactClient), fetch.WithMaxArtifactSize(s.cfg.ParseArtifactMaxSize())) fetcher := fetch.NewCircuitBreakerFetcher(baseFetcher) s.breakers = newBreakerMonitor(fetcher, s.logger) resolver := fetch.NewResolver() @@ -236,7 +237,7 @@ func (s *Server) serve(listener net.Listener) error { Ecosystems: s.cfg.Cooldown.Ecosystems, Packages: s.cfg.Cooldown.NormalizedPackages(), } - proxy := handler.NewProxy(s.db, s.storage, handler.LimitArtifactSize(fetcher, s.cfg.ParseArtifactMaxSize()), resolver, s.logger) + proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger) proxy.HTTPClient = metadataClient proxy.RateLimits = s.rateLimits cooldownPolicy, err := cooldownpolicy.New(cd, s.cfg.Cooldown.PackagePatterns) @@ -497,9 +498,24 @@ func (s *Server) newUpstreamClients(artifactTimeout time.Duration) (metadataClie } // The gate sits under the auth transport so it sees every final upstream // answer, including the retry after a bearer challenge. - gate := upstreamhttp.NewRateLimitGate(baseTransport) - gate.Logger = s.logger - gate.OnHold = metrics.RecordUpstreamRateLimitHold + gate := registryclient.NewRateLimitGate(baseTransport) + gate.OnHold = func(event registryclient.HoldEvent) { + metrics.RecordUpstreamRateLimitHold(event.Host) + if s.logger != nil { + source := "default" + if event.FromHeader { + source = "retry-after" + } + s.logger.Warn("upstream rate limited this proxy, holding its requests", + "host", event.Host, "hold", event.RetryAfter.Round(time.Second).String(), "source", source, + "until", event.Until.UTC().Format(time.RFC3339)) + } + } + gate.OnRelease = func(host string) { + if s.logger != nil { + s.logger.Info("upstream rate limit cleared", "host", host) + } + } s.rateLimits = gate authTransport := upstreamhttp.NewTransport(gate, upstreamhttp.AuthFunc(s.authForURL)) metadata := *safeClient @@ -517,7 +533,7 @@ func (s *Server) newUpstreamClients(artifactTimeout time.Duration) (metadataClie func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, logger *slog.Logger) (*handler.Proxy, *handler.ContainerHandler) { s := &Server{cfg: cfg, db: db, storage: store, logger: logger, ecr: newECRTokens(logger)} metadataClient, artifactClient := s.newUpstreamClients(0) - fetcher := handler.LimitArtifactSize(fetch.NewFetcher(fetch.WithHTTPClient(artifactClient)), cfg.ParseArtifactMaxSize()) + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(artifactClient), fetch.WithMaxArtifactSize(cfg.ParseArtifactMaxSize())) proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), logger) proxy.HTTPClient = metadataClient proxy.RateLimits = s.rateLimits