diff --git a/README.md b/README.md index 6fb3e8ab..559f1a80 100644 --- a/README.md +++ b/README.md @@ -1364,6 +1364,7 @@ The proxy exposes Prometheus metrics at `GET /metrics`. All metric names are pre | `proxy_health_probe_failures_total` | counter | `step` | Storage health probe failures by failing step (`write`, `size`, `read`, `verify`, `delete`). | | `proxy_circuit_breaker_state` | gauge | `registry` | Artifact-fetch circuit breaker state per upstream registry (0 closed, 2 open). Published once that registry's breaker has tripped. | | `proxy_circuit_breaker_trips_total` | counter | `registry` | Circuit breaker trips per upstream registry. | +| `proxy_upstream_rate_limit_holds_total` | counter | `host` | Holds placed on an upstream host after it rate limited the proxy with HTTP 429, or 503 with a `Retry-After`. See `upstream_holds` under [Health Check](#health-check). | | `proxy_ecosystem_downloaded_bytes` | gauge | `ecosystem` | Accumulated bytes served from cache: cache hits multiplied by the artifact size they served. | | `proxy_ecosystem_artifact_downloads` | gauge | `ecosystem` | Accumulated artifact downloads served from cache. | | `proxy_ecosystem_cache_size_bytes` | gauge | `ecosystem` | Size of cached artifacts per ecosystem. | @@ -1468,6 +1469,9 @@ It carries three ecosystem filters rather than one, because `ecosystem` means th "circuit_breakers": { "registry.npmjs.org": "closed", "static.crates.io": "open" + }, + "upstream_holds": { + "repo1.maven.org": "2026-10-14T11:15:29Z" } } ``` @@ -1476,6 +1480,8 @@ Failing checks include an `"error"` field. Storage failures also include a `"ste `circuit_breakers` reports the state of each upstream's artifact-fetch circuit breaker (`"open"` or `"closed"`), keyed by upstream host — or by the `hostless-url-` placeholder described under [Monitoring](#monitoring) where the fetch URL has no host to read. The key is omitted until the proxy has fetched an artifact from at least one upstream, and a host appears only once a breaker has been created for it. Breakers trip after repeated upstream failures and retry the upstream after an exponential backoff. While one is open, artifact downloads for that host return HTTP 502 on a cache miss without contacting the upstream; already-cached artifacts are still served from storage, since the cache is checked before the fetcher. A breaker is reported as `"open"` throughout its backoff, including the half-open window in which it admits one probe request to test recovery. Breaker state is per process and in memory, so a restart clears it, but a restart is not needed for recovery: the backoff keeps retrying for as long as the breaker is open, so it closes on its own once the upstream serves again. +`upstream_holds` lists upstream hosts that answered the proxy with HTTP 429, or 503 with a `Retry-After`, and the time until which the proxy sends them nothing. Every request to a held host, metadata and artifacts alike, fails locally for the duration the upstream asked for (capped at one hour), or for one minute doubling up to fifteen when the upstream gave no `Retry-After`. Cached metadata is served stale during a hold, and anything uncached returns HTTP 503 with the remaining `Retry-After` so a client can pace itself too. Metadata requests do not go through the circuit breaker, so without the hold a rate-limited upstream would keep receiving one request per cache miss, which is exactly the traffic it is asking the proxy to stop sending. The first non-limited response after a hold ends clears it, and `proxy_upstream_rate_limit_holds_total` counts the holds placed. The key is omitted while no host is held. + An open breaker does **not** set `status` to `"error"` or change the HTTP status code: it reports a specific upstream refusing to serve, not this proxy being unfit to receive traffic, and failing the readiness probe over one unhealthy upstream would pull the pod out of rotation for every other ecosystem too. Use `proxy_circuit_breaker_state` for alerting on it. Storage probe results are cached for `health.storage_probe_interval` (default 30s) to bound the cost of probing remote backends. A probe holds an internal mutex for up to 10 seconds (the hardcoded per-probe timeout), so `/health` is intended as a Kubernetes **readiness** probe rather than a liveness probe — a slow S3 round-trip should pull the pod from rotation, not restart it. diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go index 5e54d3c6..7a8387d9 100644 --- a/docs/swagger/docs.go +++ b/docs/swagger/docs.go @@ -825,6 +825,13 @@ const docTemplate = `{ }, "status": { "type": "string" + }, + "upstream_holds": { + "description": "UpstreamHolds lists upstream hosts that rate limited the proxy and\nthe RFC 3339 time until which it sends them nothing. A held host\nfails cache misses without contacting it and does not change Status.", + "type": "object", + "additionalProperties": { + "type": "string" + } } } }, diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json index 65a271b2..aad0cea0 100644 --- a/docs/swagger/swagger.json +++ b/docs/swagger/swagger.json @@ -818,6 +818,13 @@ }, "status": { "type": "string" + }, + "upstream_holds": { + "description": "UpstreamHolds lists upstream hosts that rate limited the proxy and\nthe RFC 3339 time until which it sends them nothing. A held host\nfails cache misses without contacting it and does not change Status.", + "type": "object", + "additionalProperties": { + "type": "string" + } } } }, diff --git a/internal/handler/arch.go b/internal/handler/arch.go index 6706dae4..f2abedf4 100644 --- a/internal/handler/arch.go +++ b/internal/handler/arch.go @@ -477,6 +477,9 @@ func (h *ArchHandler) writeMetadata(w http.ResponseWriter, r *http.Request, cach http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, res.err) { + return + } h.proxy.Logger.Error("metadata fetch failed", "error", res.err) http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return diff --git a/internal/handler/cargo.go b/internal/handler/cargo.go index 97d2f170..d30d2e68 100644 --- a/internal/handler/cargo.go +++ b/internal/handler/cargo.go @@ -113,6 +113,9 @@ func (h *CargoHandler) handleIndex(w http.ResponseWriter, r *http.Request) { http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch upstream index", "error", err) http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return diff --git a/internal/handler/composer.go b/internal/handler/composer.go index 9dac530c..adc4d5fb 100644 --- a/internal/handler/composer.go +++ b/internal/handler/composer.go @@ -175,6 +175,9 @@ func (h *ComposerHandler) handlePackageMetadata(w http.ResponseWriter, r *http.R http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return @@ -652,6 +655,9 @@ func (h *ComposerHandler) handleDownload(w http.ResponseWriter, r *http.Request) for _, metaURL := range metaURLs { found, err := h.findDownloadURLFromMetadata(r.Context(), metaURL, packageName, version) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch metadata", "error", err, "url", metaURL) http.Error(w, "failed to fetch metadata", http.StatusBadGateway) return @@ -890,6 +896,9 @@ func (h *ComposerHandler) proxyUpstream(w http.ResponseWriter, r *http.Request) resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/composer_hosted.go b/internal/handler/composer_hosted.go index a42a27a1..9a27e6fc 100644 --- a/internal/handler/composer_hosted.go +++ b/internal/handler/composer_hosted.go @@ -85,6 +85,9 @@ func (h *ComposerHandler) serveMetadataWithLocal(w http.ResponseWriter, r *http. case errors.Is(upstream.err, ErrUpstreamNotFound): out, err = localComposerDocument(packageName, local) case upstream.err != nil: + if writeUpstreamHeld(w, upstream.err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", upstream.err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/conan.go b/internal/handler/conan.go index 6d88b439..9bb44012 100644 --- a/internal/handler/conan.go +++ b/internal/handler/conan.go @@ -196,6 +196,9 @@ func (h *ConanHandler) proxyUpstream(w http.ResponseWriter, r *http.Request) { resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 6e4be7e9..d7b8b3f6 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -183,6 +183,9 @@ func (h *CondaHandler) handleRepodata(w http.ResponseWriter, r *http.Request) { resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/container.go b/internal/handler/container.go index 49ef3b21..55f17156 100644 --- a/internal/handler/container.go +++ b/internal/handler/container.go @@ -450,6 +450,9 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req h.containerError(w, http.StatusBadGateway, "DIGEST_INVALID", "blob digest verification failed") return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch blob", "error", err) h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch blob") return @@ -520,6 +523,9 @@ func (h *ContainerHandler) proxyBlobHead(w http.ResponseWriter, r *http.Request, resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") return } diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go index 63b2d472..902ad3aa 100644 --- a/internal/handler/container_manifest.go +++ b/internal/handler/container_manifest.go @@ -150,6 +150,9 @@ func (h *ContainerHandler) serveStaleManifestOrError(w http.ResponseWriter, r *h h.containerError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch manifest", "error", err) h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") } diff --git a/internal/handler/container_referrers.go b/internal/handler/container_referrers.go index d03e8595..37261c5d 100644 --- a/internal/handler/container_referrers.go +++ b/internal/handler/container_referrers.go @@ -160,6 +160,9 @@ func (h *ContainerHandler) serveStaleReferrersOrError(w http.ResponseWriter, r * h.writeContainerReferrers(w, r, registryURL, cached, true) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch container referrers", "error", err) h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") } diff --git a/internal/handler/container_tags.go b/internal/handler/container_tags.go index a0ac9c18..61425906 100644 --- a/internal/handler/container_tags.go +++ b/internal/handler/container_tags.go @@ -117,6 +117,9 @@ func (h *ContainerHandler) serveStaleTagsOrError(w http.ResponseWriter, r *http. h.writeContainerTags(w, r, registryURL, cached, true) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch container tag list", "error", err) h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") } diff --git a/internal/handler/download_test.go b/internal/handler/download_test.go index 2797d421..5d70ce9c 100644 --- a/internal/handler/download_test.go +++ b/internal/handler/download_test.go @@ -6,11 +6,13 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" "strings" "testing" "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" @@ -1258,3 +1260,130 @@ func TestRPMHandler_DownloadCacheMiss(t *testing.T) { t.Errorf("upstream URL = %q, want %q", fetcher.fetchedURL, want) } } + +func TestMavenHandler_UpstreamHeldReturns503(t *testing.T) { + held := &httpclient.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) + fetcher.fetchErr = fmt.Errorf("fetching artifact: %w", &url.Error{Op: "Get", URL: "https://repo1.maven.org/x", Err: held}) + + h := NewMavenHandler(proxy, "http://localhost", "", "") + srv := httptest.NewServer(h.Routes()) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/com/example/lib/1.0/lib-1.0.jar") + if err != nil { + t.Fatalf("request failed: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") != "90" { + t.Errorf("status %d Retry-After %q, want 503 with Retry-After 90", resp.StatusCode, resp.Header.Get("Retry-After")) + } + }) + + t.Run("metadata", func(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = &http.Client{Transport: roundTripperFunc(func(*http.Request) (*http.Response, error) { + return nil, held + })} + + h := NewMavenHandler(proxy, "http://localhost", "", "") + srv := httptest.NewServer(h.Routes()) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/com/example/lib/maven-metadata.xml") + if err != nil { + t.Fatalf("request failed: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") != "90" { + t.Errorf("status %d Retry-After %q, want 503 with Retry-After 90", resp.StatusCode, resp.Header.Get("Retry-After")) + } + }) +} + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { return fn(req) } + +func TestMavenHandler_HeldArtifactMissesSkipCircuitBreaker(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + upstreamCalls := 0 + gate := httpclient.NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { + upstreamCalls++ + return &http.Response{ + StatusCode: http.StatusTooManyRequests, + Header: http.Header{"Retry-After": []string{"600"}}, + Body: http.NoBody, + Request: req, + }, nil + })) + req, _ := http.NewRequest(http.MethodGet, "https://repo1.maven.org/maven2/x", nil) + if resp, err := gate.RoundTrip(req); err == nil { + _ = resp.Body.Close() + } + breaker := fetch.NewCircuitBreakerFetcher(fetch.NewFetcher(fetch.WithHTTPClient(&http.Client{Transport: gate}))) + proxy.Fetcher = breaker + proxy.RateLimits = gate + + h := NewMavenHandler(proxy, "http://localhost", "", "") + srv := httptest.NewServer(h.Routes()) + defer srv.Close() + + for i := range 8 { + resp, err := http.Get(fmt.Sprintf("%s/com/example/lib/1.%d/lib-1.%d.jar", srv.URL, i, i)) + if err != nil { + t.Fatalf("request %d failed: %v", i, err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") == "" { + t.Fatalf("request %d: status %d Retry-After %q, want 503 with Retry-After", i, resp.StatusCode, resp.Header.Get("Retry-After")) + } + } + if upstreamCalls != 1 { + t.Errorf("upstream calls = %d, want only the one that started the hold", upstreamCalls) + } + if state := breaker.GetBreakerState()["repo1.maven.org"]; state == "open" { + t.Errorf("breaker for repo1.maven.org is open after held misses, want closed") + } +} + +func TestHandlers_UpstreamHeldMetadataReturns503(t *testing.T) { + held := &httpclient.HeldError{Host: "upstream.example", Until: time.Now().Add(42 * time.Second), RetryAfter: 42 * time.Second} + tests := []struct { + name string + handler func(*Proxy) http.Handler + path string + }{ + {"npm metadata", func(p *Proxy) http.Handler { return NewNPMHandler(p, "http://localhost", "").Routes() }, "/lodash"}, + {"pypi simple", func(p *Proxy) http.Handler { return NewPyPIHandler(p, "http://localhost").Routes() }, "/simple/requests/"}, + {"pypi json", func(p *Proxy) http.Handler { return NewPyPIHandler(p, "http://localhost").Routes() }, "/pypi/requests/json"}, + {"cargo index", func(p *Proxy) http.Handler { + return NewCargoHandler(p, "http://localhost", "", "").Routes() + }, "/se/rd/serde"}, + {"pub metadata", func(p *Proxy) http.Handler { return NewPubHandler(p, "http://localhost").Routes() }, "/api/packages/http"}, + {"nuget service index", func(p *Proxy) http.Handler { return NewNuGetHandler(p, "http://localhost").Routes() }, "/v3/index.json"}, + {"nuget search", func(p *Proxy) http.Handler { return NewNuGetHandler(p, "http://localhost").Routes() }, "/query?q=json"}, + {"conda repodata", func(p *Proxy) http.Handler { return NewCondaHandler(p, "http://localhost").Routes() }, "/conda-forge/noarch/repodata.json"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = &http.Client{Transport: roundTripperFunc(func(*http.Request) (*http.Response, error) { + return nil, held + })} + srv := httptest.NewServer(tt.handler(proxy)) + defer srv.Close() + + resp, err := http.Get(srv.URL + tt.path) + if err != nil { + t.Fatalf("request failed: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") != "42" { + t.Errorf("status %d Retry-After %q, want 503 with Retry-After 42", resp.StatusCode, resp.Header.Get("Retry-After")) + } + }) + } +} diff --git a/internal/handler/gem.go b/internal/handler/gem.go index 0657b6b1..2d2136b2 100644 --- a/internal/handler/gem.go +++ b/internal/handler/gem.go @@ -121,6 +121,9 @@ func (h *GemHandler) handleCompactIndex(w http.ResponseWriter, r *http.Request) indexResp, filteredVersions, err := h.fetchIndexAndVersions(r, name) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream compact index request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return @@ -306,6 +309,9 @@ func (h *GemHandler) proxyUpstream(w http.ResponseWriter, r *http.Request) { resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/go.go b/internal/handler/go.go index a27c37be..c5284724 100644 --- a/internal/handler/go.go +++ b/internal/handler/go.go @@ -137,6 +137,9 @@ func (h *GoHandler) handleDownload(w http.ResponseWriter, r *http.Request, modul http.Error(w, err.Error(), http.StatusForbidden) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to get artifact", "error", err) http.Error(w, "failed to fetch module", http.StatusBadGateway) return diff --git a/internal/handler/go_sumdb.go b/internal/handler/go_sumdb.go index 020a126e..7de25132 100644 --- a/internal/handler/go_sumdb.go +++ b/internal/handler/go_sumdb.go @@ -407,6 +407,9 @@ func (h *GoHandler) proxySumDB(w http.ResponseWriter, r *http.Request, name, ups case errors.Is(err, ErrUpstreamNotFound): http.Error(w, "not found", http.StatusNotFound) default: + if writeUpstreamHeld(w, err) { + return + } p.Logger.Error("checksum database fetch failed", "url", upstreamURL, "error", err) http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) } diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 5e0688d8..560b8eb5 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -20,6 +20,7 @@ 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" @@ -175,6 +176,10 @@ type Proxy struct { DirectServeBaseURL string HTTPClient *http.Client + // 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 + // HostedNPMScopes and HostedComposerVendors name namespaces served from // pushed packages alone, without asking the upstream. HostedNPMScopes []string @@ -510,6 +515,9 @@ func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, fil } } + if err := p.RateLimits.Held(info.URL); err != nil { + return artifacts.Artifact{}, "", err + } p.Logger.Info("fetching from upstream", "ecosystem", ecosystem, "name", name, "version", version, "url", info.URL) @@ -1121,6 +1129,9 @@ func (p *Proxy) ProxyUpstream(w http.ResponseWriter, r *http.Request, upstreamUR resp, err := p.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } p.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return @@ -1140,6 +1151,9 @@ func (p *Proxy) ProxyFile(w http.ResponseWriter, r *http.Request, upstreamURL st resp, err := p.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return } @@ -1207,6 +1221,9 @@ func (p *Proxy) serveArtifactError(w http.ResponseWriter, err error, clientMsg s http.Error(w, "pushed package is temporarily unavailable", http.StatusServiceUnavailable) return } + if writeUpstreamHeld(w, err) { + return + } p.Logger.Error("failed to get artifact", "error", err) http.Error(w, clientMsg, http.StatusBadGateway) } @@ -1684,6 +1701,9 @@ func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } p.Logger.Error("metadata fetch failed", "error", err) http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return @@ -1770,6 +1790,9 @@ func (p *Proxy) proxyMetadataStream(w http.ResponseWriter, r *http.Request, upst resp, err := p.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return } @@ -1902,6 +1925,9 @@ func (p *Proxy) discardStaleArtifact(pkgPURL, versionPURL, filename, upstreamHas } func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL string, headers http.Header, upstreamHash, declared string) (artifacts.Artifact, string, error) { + if err := p.RateLimits.Held(downloadURL); err != nil { + return artifacts.Artifact{}, "", err + } p.Logger.Info("fetching from upstream", "ecosystem", ecosystem, "name", name, "version", version, "url", downloadURL) @@ -1929,6 +1955,9 @@ func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, versi // after the bytes have been sent, and an unterminated chunked response is the // only way left to make the client reject them (see serveArtifact). func (p *Proxy) streamFromUpstream(ctx context.Context, ecosystem, name, version, filename, versionPURL, upstreamURL, upstreamHash, declared string, fetchArtifact func(context.Context) (*fetch.Artifact, error)) (*CacheResult, error) { + if err := p.RateLimits.Held(upstreamURL); err != nil { + return nil, err + } p.Logger.Info("streaming from upstream", "ecosystem", ecosystem, "name", name, "version", version, "url", upstreamURL) diff --git a/internal/handler/helm.go b/internal/handler/helm.go index cddc9432..6183d971 100644 --- a/internal/handler/helm.go +++ b/internal/handler/helm.go @@ -186,6 +186,9 @@ func (h *HelmHandler) serveIndexError(w http.ResponseWriter, err error) { http.Error(w, "Helm repository not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch Helm index", "error", err) http.Error(w, "failed to fetch Helm repository index", http.StatusBadGateway) } diff --git a/internal/handler/hex.go b/internal/handler/hex.go index 79be510b..847f955b 100644 --- a/internal/handler/hex.go +++ b/internal/handler/hex.go @@ -118,6 +118,9 @@ func (h *HexHandler) handlePackages(w http.ResponseWriter, r *http.Request) { protoResp, filteredVersions, err := h.fetchPackageAndVersions(r, name) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/maven.go b/internal/handler/maven.go index 700c107f..027a68db 100644 --- a/internal/handler/maven.go +++ b/internal/handler/maven.go @@ -115,6 +115,9 @@ func (h *MavenHandler) handleMetadata(w http.ResponseWriter, r *http.Request, ur http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("metadata fetch failed", "error", err) http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return diff --git a/internal/handler/npm.go b/internal/handler/npm.go index 0e0d2b00..875dee7f 100644 --- a/internal/handler/npm.go +++ b/internal/handler/npm.go @@ -163,6 +163,9 @@ func (h *NPMHandler) handleSecurity(w http.ResponseWriter, r *http.Request) { resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("npm audit request failed", "path", r.URL.Path, "error", err) JSONError(w, http.StatusBadGateway, "failed to reach upstream registry") return @@ -240,6 +243,9 @@ func (h *NPMHandler) handlePackageMetadata(w http.ResponseWriter, r *http.Reques } return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch npm metadata", "error", err) JSONError(w, http.StatusBadGateway, "failed to fetch from upstream") return @@ -285,6 +291,9 @@ func (h *NPMHandler) handleFilteredNPMVersionMetadata(w http.ResponseWriter, r * JSONError(w, http.StatusNotFound, ErrPackageNotFound.Error()) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to fetch npm metadata", "error", err) JSONError(w, http.StatusBadGateway, "failed to fetch from upstream") return @@ -849,6 +858,9 @@ func (h *NPMHandler) handleDownload(w http.ResponseWriter, r *http.Request) { case errors.Is(err, ErrArtifactBlocked), errors.Is(err, ErrVersionDenied): JSONError(w, http.StatusForbidden, err.Error()) default: + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("failed to get artifact", "error", err) JSONError(w, http.StatusBadGateway, "failed to fetch package") } diff --git a/internal/handler/npm_hosted.go b/internal/handler/npm_hosted.go index e8998c90..65b59ba3 100644 --- a/internal/handler/npm_hosted.go +++ b/internal/handler/npm_hosted.go @@ -105,6 +105,9 @@ func (h *NPMHandler) serveMetadataWithLocal(w http.ResponseWriter, r *http.Reque case errors.Is(upstream.err, ErrUpstreamNotFound): out, err = localNPMPackument(packageName, local) case upstream.err != nil: + if writeUpstreamHeld(w, upstream.err) { + return + } h.proxy.Logger.Error("failed to fetch npm metadata", "error", upstream.err) JSONError(w, http.StatusBadGateway, "failed to fetch from upstream") return diff --git a/internal/handler/nuget.go b/internal/handler/nuget.go index 1a3d0a11..3d9f7278 100644 --- a/internal/handler/nuget.go +++ b/internal/handler/nuget.go @@ -96,6 +96,9 @@ func (h *NuGetHandler) handleServiceIndex(w http.ResponseWriter, r *http.Request http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return @@ -234,6 +237,9 @@ func (h *NuGetHandler) proxyUpstream(w http.ResponseWriter, r *http.Request) { resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/nuget_cooldown.go b/internal/handler/nuget_cooldown.go index 6b95ae9b..afb147d5 100644 --- a/internal/handler/nuget_cooldown.go +++ b/internal/handler/nuget_cooldown.go @@ -118,6 +118,9 @@ func (h *NuGetHandler) nugetMetadataError(w http.ResponseWriter, err error) { JSONError(w, http.StatusNotFound, "package metadata not found") return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Warn("failed to process NuGet metadata", "error", err) JSONError(w, http.StatusBadGateway, "failed to process package metadata") } diff --git a/internal/handler/pub.go b/internal/handler/pub.go index e907e737..12ff6d92 100644 --- a/internal/handler/pub.go +++ b/internal/handler/pub.go @@ -111,6 +111,9 @@ func (h *PubHandler) handlePackageMetadata(w http.ResponseWriter, r *http.Reques http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/pypi.go b/internal/handler/pypi.go index f4cef06a..513f5077 100644 --- a/internal/handler/pypi.go +++ b/internal/handler/pypi.go @@ -118,6 +118,9 @@ func (h *PyPIHandler) handleSimplePackage(w http.ResponseWriter, r *http.Request http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return @@ -418,6 +421,9 @@ func (h *PyPIHandler) proxyAndRewriteJSON(w http.ResponseWriter, r *http.Request http.Error(w, "not found", http.StatusNotFound) return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return @@ -821,6 +827,9 @@ func (h *PyPIHandler) proxySimple(w http.ResponseWriter, r *http.Request, path s resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("upstream request failed", "error", err) http.Error(w, "upstream request failed", http.StatusBadGateway) return diff --git a/internal/handler/swift.go b/internal/handler/swift.go index 243a37cc..b506f1c2 100644 --- a/internal/handler/swift.go +++ b/internal/handler/swift.go @@ -439,6 +439,9 @@ func (h *SwiftHandler) proxySwiftResource(w http.ResponseWriter, r *http.Request resp, err := h.proxy.HTTPClient.Do(req) if err != nil { + if writeUpstreamHeld(w, err) { + return + } writeSwiftProblem(w, http.StatusBadGateway, "upstream request failed") return } @@ -613,6 +616,9 @@ func (h *SwiftHandler) writeMetadataError(w http.ResponseWriter, err error) { writeSwiftProblem(w, http.StatusNotFound, "not found") return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("Swift metadata request failed", "error", err) writeSwiftProblem(w, http.StatusBadGateway, "upstream request failed") } @@ -622,6 +628,9 @@ func (h *SwiftHandler) writeArtifactError(w http.ResponseWriter, err error) { writeSwiftProblem(w, http.StatusNotFound, "release not found") return } + if writeUpstreamHeld(w, err) { + return + } h.proxy.Logger.Error("Swift archive request failed", "error", err) writeSwiftProblem(w, http.StatusBadGateway, "failed to fetch package") } diff --git a/internal/handler/upstream_held.go b/internal/handler/upstream_held.go new file mode 100644 index 00000000..fa2ff968 --- /dev/null +++ b/internal/handler/upstream_held.go @@ -0,0 +1,27 @@ +package handler + +import ( + "errors" + "math" + "net/http" + "strconv" + + "github.com/git-pkgs/proxy/internal/httpclient" +) + +// 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 + if !errors.As(err, &held) { + return false + } + secs := int(math.Ceil(held.RetryAfter.Seconds())) + if secs < 1 { + secs = 1 + } + w.Header().Set("Retry-After", strconv.Itoa(secs)) + http.Error(w, held.Error(), http.StatusServiceUnavailable) + return true +} diff --git a/internal/httpclient/ratelimit.go b/internal/httpclient/ratelimit.go new file mode 100644 index 00000000..53e3e8ef --- /dev/null +++ b/internal/httpclient/ratelimit.go @@ -0,0 +1,228 @@ +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 new file mode 100644 index 00000000..8ad7e9c4 --- /dev/null +++ b/internal/httpclient/ratelimit_test.go @@ -0,0 +1,257 @@ +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/metrics/metrics.go b/internal/metrics/metrics.go index 7567ff8e..5e9e1a64 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -105,6 +105,14 @@ var ( []string{"registry"}, ) + UpstreamRateLimitHolds = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "proxy_upstream_rate_limit_holds_total", + Help: "Holds placed on an upstream host after it rate limited the proxy", + }, + []string{"host"}, + ) + // Storage metrics StorageOperationDuration = prometheus.NewHistogramVec( prometheus.HistogramOpts{ @@ -263,6 +271,7 @@ func init() { UpstreamErrors, CircuitBreakerState, CircuitBreakerTrips, + UpstreamRateLimitHolds, StorageOperationDuration, StorageErrors, ActiveRequests, @@ -482,6 +491,11 @@ func RecordCircuitBreakerTrip(registry string) { CircuitBreakerTrips.WithLabelValues(registry).Inc() } +// RecordUpstreamRateLimitHold counts a hold placed on host. +func RecordUpstreamRateLimitHold(host string) { + UpstreamRateLimitHolds.WithLabelValues(host).Inc() +} + // IncrementActiveRequests increments the active request counter. func IncrementActiveRequests() { ActiveRequests.Inc() diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index cca39c89..8259d3e3 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -88,6 +88,7 @@ func TestMetricsAreRegistered(t *testing.T) { CachedArtifacts, CircuitBreakerState, CircuitBreakerTrips, + UpstreamRateLimitHolds, ActiveRequests, } diff --git a/internal/server/breakers_test.go b/internal/server/breakers_test.go index ae397596..bb5166ee 100644 --- a/internal/server/breakers_test.go +++ b/internal/server/breakers_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + upstreamhttp "github.com/git-pkgs/proxy/internal/httpclient" "github.com/git-pkgs/proxy/internal/metrics" "github.com/git-pkgs/registries/fetch" "github.com/prometheus/client_golang/prometheus" @@ -346,3 +347,44 @@ func TestBreakerMonitor_SnapshotSerializesStateReads(t *testing.T) { t.Errorf("trips = %v, want 1", trips) } } + +func TestHealthEndpoint_ReportsUpstreamHolds(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + + gate := upstreamhttp.NewRateLimitGate(roundTripperFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusTooManyRequests, + Header: http.Header{"Retry-After": []string{"600"}}, + Body: http.NoBody, + Request: req, + }, nil + })) + req, _ := http.NewRequest(http.MethodGet, "https://repo1.maven.org/maven2/x", nil) + if resp, err := gate.RoundTrip(req); err == nil { + _ = resp.Body.Close() + } + ts.server.rateLimits = gate + + w := httptest.NewRecorder() + ts.handler.ServeHTTP(w, httptest.NewRequest("GET", "/health", nil)) + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String()) + } + + var resp HealthResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decoding response: %v", err) + } + until, err := time.Parse(time.RFC3339, resp.UpstreamHolds["repo1.maven.org"]) + if err != nil { + t.Fatalf("upstream_holds = %v, want repo1.maven.org with an RFC 3339 deadline", resp.UpstreamHolds) + } + if remaining := time.Until(until); remaining < 9*time.Minute || remaining > 10*time.Minute { + t.Errorf("hold ends in %s, want about 10m", remaining) + } +} + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { return fn(req) } diff --git a/internal/server/health.go b/internal/server/health.go index fd49cf13..cc1236db 100644 --- a/internal/server/health.go +++ b/internal/server/health.go @@ -38,6 +38,22 @@ type HealthResponse struct { // the upstream, but says nothing about this proxy's own health, so it does // not change Status. CircuitBreakers map[string]string `json:"circuit_breakers,omitempty"` + // UpstreamHolds lists upstream hosts that rate limited the proxy and + // the RFC 3339 time until which it sends them nothing. A held host + // fails cache misses without contacting it and does not change Status. + UpstreamHolds map[string]string `json:"upstream_holds,omitempty"` +} + +// formatHolds renders hold deadlines for the health report; nil when none. +func formatHolds(holds map[string]time.Time) map[string]string { + if len(holds) == 0 { + return nil + } + out := make(map[string]string, len(holds)) + for host, until := range holds { + out[host] = until.UTC().Format(time.RFC3339) + } + return out } // HealthCheck reports the status of a single subsystem check. diff --git a/internal/server/server.go b/internal/server/server.go index 5b35435c..c503e28f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -117,6 +117,7 @@ type Server struct { accessLog *accesslog.Logger ecr *ecrTokens breakers *breakerMonitor + rateLimits *upstreamhttp.RateLimitGate ecoStats ecosystemStatsCache sources sourceTracker @@ -237,6 +238,7 @@ func (s *Server) serve(listener net.Listener) error { } proxy := handler.NewProxy(s.db, s.storage, handler.LimitArtifactSize(fetcher, s.cfg.ParseArtifactMaxSize()), resolver, s.logger) proxy.HTTPClient = metadataClient + proxy.RateLimits = s.rateLimits cooldownPolicy, err := cooldownpolicy.New(cd, s.cfg.Cooldown.PackagePatterns) if err != nil { return fmt.Errorf("configuring cooldown policy: %w", err) @@ -493,7 +495,13 @@ func (s *Server) newUpstreamClients(artifactTimeout time.Duration) (metadataClie if s.accessLog != nil { baseTransport = upstreamhttp.NewAccessLogTransport(baseTransport, s.accessLog, s.logger) } - authTransport := upstreamhttp.NewTransport(baseTransport, upstreamhttp.AuthFunc(s.authForURL)) + // 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 + s.rateLimits = gate + authTransport := upstreamhttp.NewTransport(gate, upstreamhttp.AuthFunc(s.authForURL)) metadata := *safeClient metadata.Timeout = s.cfg.ParseHTTPTimeout() metadata.Transport = authTransport @@ -512,6 +520,7 @@ func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, fetcher := handler.LimitArtifactSize(fetch.NewFetcher(fetch.WithHTTPClient(artifactClient)), cfg.ParseArtifactMaxSize()) proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), logger) proxy.HTTPClient = metadataClient + proxy.RateLimits = s.rateLimits return proxy, s.newContainerHandler(proxy) } @@ -1156,6 +1165,7 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { Status: "ok", Checks: map[string]HealthCheck{}, CircuitBreakers: s.breakers.snapshot(), + UpstreamHolds: formatHolds(s.rateLimits.Holds()), } // Database check (short-circuit; do not waste a storage probe call when DB is down).