Hold requests to an upstream that rate limits the proxy

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-<digest>` 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.

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"
+                    }
                 }
             }
         },

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"
+                    }
                 }
             }
         },

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

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

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

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

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

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

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
 	}

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")
 }

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")
 }

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")
 }

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"))
+			}
+		})
+	}
+}

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

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

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)
 		}