fix: scope upstream cooldowns and exempt internal forwards

README.md

@@ -219,14 +219,15 @@ spool {
 
 ## Upstream Rate Limits
 
-After an upstream returns 429, cachewd pauses HTTP requests to that outbound hostname across strategies, schemes, and ports until a cooldown ends. Each redirect destination is checked independently. Cache hits and other hostnames remain available. OCI blob, manifest, authentication, and pass-through requests use the same shared cooldowns.
+After an upstream returns 429, cachewd pauses HTTP requests to that outbound hostname across strategies, schemes, and ports until a cooldown ends. Requests made with a GitHub App installation token, an App JWT, or the static GitHub token cool down separately, because GitHub limits each of those credentials on its own: one installation's 429 does not pause another installation, token minting, or anonymous requests. All other requests share their hostname's cooldown. A redirect to another host drops the credential, so that hop also shares the hostname's cooldown. Each redirect destination is checked independently. Cache hits and other hostnames remain available. Requests that one strategy forwards to another cachew route, such as Hermit's GitHub releases, are exempt; the target route applies the cooldown to the real upstream. OCI blob, manifest, authentication, and pass-through requests use the same shared cooldowns.
 
 Configured fallbacks are tried on the first 429 and on subsequent cache misses while the primary is cooling. A mirror on another hostname remains available until it returns its own 429; its 404 or 5xx does not start a cooldown. Requests to cooling hosts get a local 429 with `Retry-After` and `Cache-Control: no-store`. Cooldowns do not schedule retries.
 
-The cooldown uses the upstream's `Retry-After` (seconds or HTTP date) when it is valid, and `upstream-backoff` otherwise. Cooldowns are held in memory per process and apply to clients that use the shared HTTP transport, not subprocess traffic. Other replicas and requests already in flight are unaffected.
+The cooldown uses the upstream's `Retry-After` (seconds or HTTP date) when it is valid, and `upstream-backoff` otherwise, and never lasts longer than `upstream-backoff-max`. Each cooldown that starts or is extended is logged and counted in `cachew.upstream.cooldowns`; local 429s are counted in `cachew.upstream.cooldown_rejections`. Cooldowns are held in memory per process and apply to clients that use the shared HTTP transport, not subprocess traffic. Other replicas and requests already in flight are unaffected.
 
 ```hcl
-upstream-backoff = "1m"  # Must be positive
+upstream-backoff = "1m"       # Must be positive
+upstream-backoff-max = "15m"  # Raised to upstream-backoff when lower
 ```
 
 ## Authorization (OPA)

cmd/cachewd/main.go

@@ -48,6 +48,7 @@ type GlobalConfig struct {
 	Bind                   string        `hcl:"bind" default:"127.0.0.1:8080" help:"Bind address for the server."`
 	URL                    string        `hcl:"url" default:"http://127.0.0.1:8080/" help:"Base URL for cachewd."`
 	UpstreamBackoff        time.Duration `hcl:"upstream-backoff,optional" default:"1m" help:"Cooldown after an upstream 429 with missing or invalid Retry-After; must be positive."`
+	UpstreamBackoffMax     time.Duration `hcl:"upstream-backoff-max,optional" default:"15m" help:"Longest cooldown after an upstream 429, whatever its Retry-After; raised to upstream-backoff when lower."`
 	ShutdownReadinessDelay time.Duration `hcl:"shutdown-readiness-delay,optional" default:"5s" help:"Delay between flipping readiness to 503 on SIGTERM and starting graceful shutdown."`
 	// ShutdownTimeout must be less than the pod's terminationGracePeriodSeconds
 	// (minus ShutdownReadinessDelay) or the kubelet will SIGKILL before Shutdown returns.
@@ -105,7 +106,7 @@ func main() {
 	// Flipped on SIGTERM so /_readiness fails before the listener closes.
 	var shuttingDown atomic.Bool
 
-	configureProcess(cli.MutexProfileFraction, globalConfig.UpstreamBackoff)
+	configureProcess(logger, cli.MutexProfileFraction, globalConfig.UpstreamBackoff, globalConfig.UpstreamBackoffMax)
 
 	// Register the OpenTelemetry tracer provider. tracing.New is a
 	// no-op when OTEL_EXPORTER_OTLP_ENDPOINT is unset, so deployments
@@ -449,17 +450,17 @@ func newServer(
 }
 
 // configureProcess applies process-wide runtime and HTTP transport settings.
-func configureProcess(mutexProfileFraction int, upstreamBackoff time.Duration) {
+func configureProcess(logger *slog.Logger, mutexProfileFraction int, upstreamBackoff, upstreamBackoffMax time.Duration) {
 	// Enable mutex contention sampling so /admin/pprof/mutex returns
 	// useful data. The pprof endpoints themselves are mounted on the
 	// main HTTP listener under /admin/pprof/* via the net/http/pprof
 	// side-effect import; OPA gates external access.
 	runtime.SetMutexProfileFraction(mutexProfileFraction)
 
 	// Every strategy that uses http.DefaultTransport shares one set of
-	// per-hostname 429 cooldowns. Origin metrics sit below the cooldowns so that
+	// per-hostname and per-credential 429 cooldowns. Origin metrics sit below the cooldowns so that
 	// requests rejected during a cooldown are not counted as origin traffic.
-	http.DefaultTransport = httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff) //nolint:reassign
+	http.DefaultTransport = httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff, upstreamBackoffMax, logger) //nolint:reassign
 }
 
 // loadGlobalConfig unmarshals the global config from HCL, using a two-pass
@@ -505,6 +506,8 @@ func loadGlobalConfig(ast *hcl.AST) (GlobalConfig, map[string]string, error) {
 	if cfg.UpstreamBackoff <= 0 {
 		return cfg, nil, errors.New("upstream-backoff must be positive")
 	}
+	// Keeps configs that set only a long upstream-backoff valid.
+	cfg.UpstreamBackoffMax = max(cfg.UpstreamBackoffMax, cfg.UpstreamBackoff)
 
 	return cfg, envars, nil
 }

internal/githubapp/ratelimit.go

@@ -98,7 +98,7 @@ func (p *RateLimitPoller) poll(ctx context.Context) error {
 func (p *RateLimitPoller) pollInstallation(ctx context.Context, app *appState, installationID string) error {
 	ctx, cancel := context.WithTimeout(ctx, rateLimitPollTimeout)
 	defer cancel()
-	token, owner, err := app.getInstallationToken(ctx, installationID)
+	token, installation, err := app.getInstallationToken(ctx, installationID)
 	if err != nil {
 		return errors.Wrap(err, "get installation token")
 	}
@@ -107,9 +107,10 @@ func (p *RateLimitPoller) pollInstallation(ctx context.Context, app *appState, i
 		return errors.Wrap(err, "create rate-limit request")
 	}
 	req.Header.Set("Authorization", "Bearer "+token)
+	req = req.WithContext(WithInstallationCooldown(req.Context(), installationID))
 	req.Header.Set("Accept", "application/vnd.github+json")
 	req.Header.Set("X-Github-Api-Version", "2022-11-28")
-	resp, err := app.httpClient.Do(WithAPIMetrics(req, owner, "get_rate_limit"))
+	resp, err := app.httpClient.Do(WithAPIMetrics(req, installation.Owner, "get_rate_limit"))
 	if err != nil {
 		return errors.Wrap(err, "fetch rate limit")
 	}
@@ -126,7 +127,7 @@ func (p *RateLimitPoller) pollInstallation(ctx context.Context, app *appState, i
 	for resource, values := range result.Resources {
 		attrs := metric.WithAttributes(
 			attribute.String("app", app.appID), attribute.String("installation_id", installationID),
-			attribute.String("owner", owner), attribute.String("resource", resource),
+			attribute.String("owner", installation.Owner), attribute.String("resource", resource),
 		)
 		p.remaining.Record(ctx, values.Remaining, attrs)
 		p.limit.Record(ctx, values.Limit, attrs)

internal/githubapp/ratelimit_test.go

@@ -161,10 +161,10 @@ func TestRateLimitPollerRetainsOwnerAcrossAliases(t *testing.T) {
 	poller := NewRateLimitPoller(&TokenManager{apps: []*appState{app}}, 0)
 	for _, alias := range []string{"Example-Org", "old-name", "renamed-org"} {
 		app.rememberInstallation(alias, "123", alias)
-		token, owner, err := app.getToken(pollerContext(t), alias)
+		token, installation, err := app.getToken(pollerContext(t), alias)
 		assert.NoError(t, err)
 		assert.Equal(t, "token-111", token)
-		assert.Equal(t, "example-org", owner)
+		assert.Equal(t, Installation{ID: "123", Owner: "example-org"}, installation)
 		assert.Equal(t, map[string]string{"123": "example-org"}, app.knownInstallations())
 		assert.NoError(t, poller.poll(pollerContext(t)))
 		points := collectMetrics(t, reader)["github.ratelimit_remaining"]

internal/githubapp/tokens.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/alecthomas/errors"
 
+	"github.com/block/cachew/internal/httputil"
 	"github.com/block/cachew/internal/logging"
 )
 
@@ -110,14 +111,21 @@ func newTokenManager(configs []Config, logger *slog.Logger, apiBase string, http
 // caches the result, and falls back to the fallback org's token for orgs where
 // the app is not installed.
 func (tm *TokenManager) GetTokenForOrg(ctx context.Context, org string) (string, error) {
-	token, _, err := tm.GetTokenForOrgWithOwner(ctx, org)
+	token, _, err := tm.GetTokenForOrgWithInstallation(ctx, org)
 	return token, err
 }
 
-// GetTokenForOrgWithOwner also returns the owner whose installation minted the token.
-func (tm *TokenManager) GetTokenForOrgWithOwner(ctx context.Context, org string) (string, string, error) {
+// Installation is the GitHub App installation that minted a token.
+type Installation struct {
+	// ID is unique across GitHub Apps.
+	ID    string
+	Owner string
+}
+
+// GetTokenForOrgWithInstallation also returns the installation that minted the token.
+func (tm *TokenManager) GetTokenForOrgWithInstallation(ctx context.Context, org string) (string, Installation, error) {
 	if tm == nil {
-		return "", "", errors.New("token manager not initialized")
+		return "", Installation{}, errors.New("token manager not initialized")
 	}
 
 	logger := logging.FromContext(ctx)
@@ -151,12 +159,12 @@ func (tm *TokenManager) GetTokenForOrgWithOwner(ctx context.Context, org string)
 	if tm.fallbackApp != nil {
 		logger.InfoContext(ctx, "Using fallback org token", "requested_org", org, "fallback_org", tm.fallbackOrg)
 		if _, err := tm.fallbackApp.lookupInstallationID(ctx, tm.fallbackOrg); err != nil {
-			return "", "", errors.Wrap(err, "resolve fallback installation")
+			return "", Installation{}, errors.Wrap(err, "resolve fallback installation")
 		}
 		return tm.fallbackApp.getToken(ctx, tm.fallbackOrg)
 	}
 
-	return "", "", errors.Errorf("no GitHub App installation found for org: %s", org)
+	return "", Installation{}, errors.Errorf("no GitHub App installation found for org: %s", org)
 }
 
 // GetTokenForURL extracts the org from a GitHub URL and returns an installation token.
@@ -172,23 +180,24 @@ func (tm *TokenManager) GetTokenForURL(ctx context.Context, url string) (string,
 	return tm.GetTokenForOrg(ctx, org)
 }
 
-func (a *appState) getToken(ctx context.Context, org string) (string, string, error) {
+func (a *appState) getToken(ctx context.Context, org string) (string, Installation, error) {
 	a.installationMu.RLock()
 	installationID := a.installationCache[org]
 	a.installationMu.RUnlock()
 	if installationID == "" {
-		return "", "", errors.Errorf("no installation ID for org: %s", org)
+		return "", Installation{}, errors.Errorf("no installation ID for org: %s", org)
 	}
 	return a.getInstallationToken(ctx, installationID)
 }
 
-func (a *appState) getInstallationToken(ctx context.Context, installationID string) (string, string, error) {
+func (a *appState) getInstallationToken(ctx context.Context, installationID string) (string, Installation, error) {
 	a.installationMu.RLock()
 	owner := a.installationOwners[installationID]
 	a.installationMu.RUnlock()
 	if owner == "" {
-		return "", "", errors.Errorf("no owner for installation: %s", installationID)
+		return "", Installation{}, errors.Errorf("no owner for installation: %s", installationID)
 	}
+	installation := Installation{ID: installationID, Owner: owner}
 	logger := logging.FromContext(ctx).With("org", owner, "app_id", a.appID)
 
 	a.mu.RLock()
@@ -197,14 +206,14 @@ func (a *appState) getInstallationToken(ctx context.Context, installationID stri
 
 	if exists && time.Now().Add(a.cacheConfig.RefreshBuffer).Before(cached.expiresAt) {
 		logger.DebugContext(ctx, "Using cached GitHub App token")
-		return cached.token, owner, nil
+		return cached.token, installation, nil
 	}
 
 	logger.DebugContext(ctx, "Fetching new GitHub App installation token", "installation_id", installationID)
 
 	token, expiresAt, err := a.fetchInstallationToken(ctx, owner, installationID)
 	if err != nil {
-		return "", "", errors.Wrap(err, "fetch installation token")
+		return "", Installation{}, errors.Wrap(err, "fetch installation token")
 	}
 
 	a.mu.Lock()
@@ -216,7 +225,7 @@ func (a *appState) getInstallationToken(ctx context.Context, installationID stri
 
 	logger.InfoContext(ctx, "GitHub App token refreshed", "expires_at", expiresAt)
 
-	return token, owner, nil
+	return token, installation, nil
 }
 
 func (a *appState) fetchInstallationToken(ctx context.Context, org, installationID string) (string, time.Time, error) {
@@ -233,6 +242,7 @@ func (a *appState) fetchInstallationToken(ctx context.Context, org, installation
 
 	req.Header.Set("Accept", "application/vnd.github+json")
 	req.Header.Set("Authorization", "Bearer "+jwt)
+	req = req.WithContext(httputil.WithCooldownScope(req.Context(), "github-app:"+a.appID))
 	req.Header.Set("X-Github-Api-Version", "2022-11-28")
 
 	resp, err := a.httpClient.Do(WithAPIMetrics(req, org, "create_installation_token"))
@@ -285,6 +295,7 @@ func (a *appState) lookupInstallationID(ctx context.Context, org string) (string
 
 		req.Header.Set("Accept", "application/vnd.github+json")
 		req.Header.Set("Authorization", "Bearer "+jwt)
+		req = req.WithContext(httputil.WithCooldownScope(req.Context(), "github-app:"+a.appID))
 		req.Header.Set("X-Github-Api-Version", "2022-11-28")
 
 		resp, err := a.httpClient.Do(WithAPIMetrics(req, "unknown", "lookup_installation"))

internal/githubapp/tokens_test.go

@@ -149,17 +149,17 @@ func TestGetTokenForOrgDynamicDiscovery(t *testing.T) {
 	ctx := logging.ContextWithLogger(t.Context(), slog.Default())
 
 	t.Run("FallbackBeforeDiscovery", func(t *testing.T) {
-		token, owner, err := tm.GetTokenForOrgWithOwner(ctx, "unknown-org")
+		token, installation, err := tm.GetTokenForOrgWithInstallation(ctx, "unknown-org")
 		assert.NoError(t, err)
 		assert.Equal(t, "ghs_squareup_token", token)
-		assert.Equal(t, "squareup", owner)
+		assert.Equal(t, githubapp.Installation{ID: "12345", Owner: "squareup"}, installation)
 	})
 
 	t.Run("DiscoverAndCacheInstallation", func(t *testing.T) {
-		token, owner, err := tm.GetTokenForOrgWithOwner(ctx, "squareup")
+		token, installation, err := tm.GetTokenForOrgWithInstallation(ctx, "squareup")
 		assert.NoError(t, err)
 		assert.Equal(t, "ghs_squareup_token", token)
-		assert.Equal(t, "squareup", owner)
+		assert.Equal(t, githubapp.Installation{ID: "12345", Owner: "squareup"}, installation)
 
 		// Second call should use cache
 		token, err = tm.GetTokenForOrg(ctx, "squareup")
@@ -168,10 +168,10 @@ func TestGetTokenForOrgDynamicDiscovery(t *testing.T) {
 	})
 
 	t.Run("DiscoverNewOrg", func(t *testing.T) {
-		token, owner, err := tm.GetTokenForOrgWithOwner(ctx, "AfterpayTouch")
+		token, installation, err := tm.GetTokenForOrgWithInstallation(ctx, "AfterpayTouch")
 		assert.NoError(t, err)
 		assert.Equal(t, "ghs_afterpay_token", token)
-		assert.Equal(t, "AfterpayTouch", owner)
+		assert.Equal(t, githubapp.Installation{ID: "67890", Owner: "AfterpayTouch"}, installation)
 	})
 
 	t.Run("FallbackForUnknownOrg", func(t *testing.T) {
@@ -230,10 +230,10 @@ func TestGetTokenForOrgRetainsInstallationOwner(t *testing.T) {
 		slog.Default(), server.URL, server.Client())
 	ctx := logging.ContextWithLogger(t.Context(), slog.Default())
 	for _, alias := range []string{"Example-Org", "example-org", "renamed-org"} {
-		token, owner, err := manager.GetTokenForOrgWithOwner(ctx, alias)
+		token, installation, err := manager.GetTokenForOrgWithInstallation(ctx, alias)
 		assert.NoError(t, err)
 		assert.Equal(t, "test-token", token)
-		assert.Equal(t, "example-org", owner)
+		assert.Equal(t, githubapp.Installation{ID: "123", Owner: "example-org"}, installation)
 	}
 	assert.Equal(t, int64(1), mints.Load())
 }

internal/githubapp/transport.go

@@ -10,6 +10,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 
+	"github.com/block/cachew/internal/httputil"
 	"github.com/block/cachew/internal/logging"
 	"github.com/block/cachew/internal/metrics"
 )
@@ -42,6 +43,12 @@ func NewAPIClient(base *http.Client) *http.Client {
 	return &client
 }
 
+// WithInstallationCooldown scopes upstream cooldowns to a GitHub App
+// installation, since GitHub limits each installation separately.
+func WithInstallationCooldown(ctx context.Context, installationID string) context.Context {
+	return httputil.WithCooldownScope(ctx, "github-installation:"+installationID)
+}
+
 // WithAPIMetrics attributes an API request to its credential owner and operation.
 func WithAPIMetrics(req *http.Request, owner, operation string) *http.Request {
 	call := apiCall{host: req.URL.Host, owner: owner, operation: operation}