Share upstream rate-limit holds across instances

README.md

@@ -1630,7 +1630,7 @@ 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.
+`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. Holds are recorded in the database, so instances sharing one honour each other's holds and list them here too. 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.
 

docs/configuration.md

@@ -207,7 +207,9 @@ Instances coordinate through PostgreSQL advisory locks:
 - Eviction, reclaim, retention and the Gradle build cache sweep run on one instance at a time. An instance that finds a job running elsewhere skips that pass.
 - A [scheduled mirror](#scheduled-refresh) runs on whichever instance reaches it first in each interval.
 
-Cache misses are fetched from upstream once across instances. The instance that fetches holds a lease on the miss in the `fetch_leases` table and renews it while it downloads. Another instance missing the same file waits until it is stored and serves the stored copy; `proxy_fetch_lease_waits_total` counts those waits. If the fetch fails, the instances waiting on it stop waiting together: each serves its stale metadata copy if it has one, and otherwise fetches itself, so a failing upstream costs every instance one attempt at the same time rather than one after another. A lease of an instance that stopped expires after 30 seconds. Metadata is shared this way only with [`cache_metadata`](#metadata-caching) and a `metadata_ttl` above zero, since without a stored copy there is nothing to wait for. A cache hit costs no extra query; a miss costs one.
+Cache misses are fetched from upstream once across instances. The instance that fetches holds a lease on the miss in the `fetch_leases` table and renews it while it downloads. Another instance missing the same file waits until it is stored and serves the stored copy; `proxy_fetch_lease_waits_total` counts those waits. If the fetch fails, the instances waiting on it stop waiting together: each serves its stale metadata copy if it has one, and otherwise fetches itself, so a failing upstream costs every instance one attempt at the same time rather than one after another. When the fetch failed because the upstream rate limited it, the waiters without a stale copy answer 503 with the upstream's `Retry-After` instead of fetching. A lease of an instance that stopped expires after 30 seconds. Metadata is shared this way only with [`cache_metadata`](#metadata-caching) and a `metadata_ttl` above zero, since without a stored copy there is nothing to wait for. A cache hit costs no extra query; a miss costs one.
+
+Rate-limit holds (see `upstream_holds` under [Health Check](../README.md#health-check)) are shared through the `upstream_holds` table. When one instance gets a 429, or a 503 with `Retry-After`, it records the host and the end of the hold, and the other instances stop sending to that host too instead of each waiting for its own 429. Each instance rereads the table at most every two seconds, so another instance can still send for up to two seconds after a hold starts. A hold is also recorded with SQLite, where it then outlasts a restart.
 
 Each instance needs its own [`git.cache_dir`](#git-repositories).
 

internal/database/database_test.go

@@ -613,7 +613,7 @@ func createTestPostgresDB(t *testing.T) *DB {
 	// Drop and recreate every table CreateSchema creates for clean test state;
 	// leftover migration records make the next CreateSchema fail on the
 	// migrations primary key.
-	tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "job_runs", "fetch_leases", "migrations", "schema_info"}
+	tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "job_runs", "fetch_leases", "upstream_holds", "migrations", "schema_info"}
 	for _, table := range tables {
 		_, _ = db.Exec("DROP TABLE IF EXISTS " + table + " CASCADE")
 	}

internal/database/schema.go

@@ -147,6 +147,11 @@ CREATE TABLE IF NOT EXISTS metadata_dates (
 	PRIMARY KEY (ecosystem, proxy_url, name)
 );
 
+CREATE TABLE IF NOT EXISTS upstream_holds (
+	host TEXT NOT NULL PRIMARY KEY,
+	until_ms INTEGER NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS migrations (
 	name TEXT NOT NULL PRIMARY KEY,
 	applied_at DATETIME NOT NULL
@@ -283,6 +288,11 @@ CREATE TABLE IF NOT EXISTS metadata_dates (
 	PRIMARY KEY (ecosystem, proxy_url, name)
 );
 
+CREATE TABLE IF NOT EXISTS upstream_holds (
+	host TEXT NOT NULL PRIMARY KEY,
+	until_ms BIGINT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS migrations (
 	name TEXT NOT NULL PRIMARY KEY,
 	applied_at TIMESTAMP NOT NULL
@@ -447,6 +457,7 @@ var migrations = []migration{
 	{"014_add_metadata_last_accessed", migrateAddMetadataLastAccessed},
 	{"015_add_job_runs", migrateAddJobRuns},
 	{"016_add_fetch_leases", migrateAddFetchLeases},
+	{"018_add_upstream_holds", migrateAddUpstreamHolds},
 }
 
 // isTableNotFound returns true if the error indicates a missing table.
@@ -782,6 +793,23 @@ func migrateAddFetchLeases(db *DB) error {
 	return nil
 }
 
+// migrateAddUpstreamHolds creates the table through which instances sharing
+// the database honour each other's upstream rate-limit holds.
+func migrateAddUpstreamHolds(db *DB) error {
+	millis := "INTEGER"
+	if db.dialect == DialectPostgres {
+		millis = "BIGINT"
+	}
+	query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS upstream_holds (
+		host TEXT NOT NULL PRIMARY KEY,
+		until_ms %s NOT NULL
+	)`, millis)
+	if _, err := db.Exec(query); err != nil {
+		return fmt.Errorf("creating upstream_holds table: %w", err)
+	}
+	return nil
+}
+
 // migrateAddPendingDeletesQueuedAtIndex indexes the column reclaim orders
 // the queue by, which grows large after a retention sweep.
 func migrateAddPendingDeletesQueuedAtIndex(db *DB) error {

internal/database/upstream_holds.go

@@ -0,0 +1,58 @@
+package database
+
+import (
+	"fmt"
+	"time"
+)
+
+// Upstream holds let instances sharing a database honour a rate-limit hold
+// any one of them placed, so each does not need its own 429 before it stops
+// sending. Times are the instances' own clocks in Unix milliseconds; a few
+// seconds of skew does not matter next to holds of a minute or more.
+
+type upstreamHoldRow struct {
+	Host    string `db:"host"`
+	UntilMS int64  `db:"until_ms"`
+}
+
+// PutUpstreamHold records that requests to host are held until until. A hold
+// already recorded for longer is kept.
+func (db *DB) PutUpstreamHold(host string, until time.Time) error {
+	_, err := db.Exec(db.Rebind(`
+		INSERT INTO upstream_holds (host, until_ms) VALUES (?, ?)
+		ON CONFLICT (host) DO UPDATE SET until_ms = CASE
+			WHEN excluded.until_ms > upstream_holds.until_ms THEN excluded.until_ms
+			ELSE upstream_holds.until_ms END
+	`), host, until.UnixMilli())
+	if err != nil {
+		return fmt.Errorf("recording upstream hold: %w", err)
+	}
+	return nil
+}
+
+// UpstreamHolds returns the holds that have not ended by now, and deletes
+// those that have.
+func (db *DB) UpstreamHolds(now time.Time) (map[string]time.Time, error) {
+	var rows []upstreamHoldRow
+	if err := db.Select(&rows, `SELECT host, until_ms FROM upstream_holds`); err != nil {
+		return nil, fmt.Errorf("reading upstream holds: %w", err)
+	}
+	holds := make(map[string]time.Time, len(rows))
+	expired := false
+	for _, row := range rows {
+		until := time.UnixMilli(row.UntilMS)
+		if now.Before(until) {
+			holds[row.Host] = until
+		} else {
+			expired = true
+		}
+	}
+	// Only write when there is something to delete, so the frequent reads
+	// stay reads.
+	if expired {
+		if _, err := db.Exec(db.Rebind(`DELETE FROM upstream_holds WHERE until_ms <= ?`), now.UnixMilli()); err != nil {
+			return holds, fmt.Errorf("deleting expired upstream holds: %w", err)
+		}
+	}
+	return holds, nil
+}

internal/database/upstream_holds_test.go

@@ -0,0 +1,48 @@
+package database
+
+import (
+	"testing"
+	"time"
+)
+
+func TestUpstreamHolds(t *testing.T) {
+	runWithBothDatabases(t, func(t *testing.T, db *DB) {
+		now := time.UnixMilli(time.Now().UnixMilli())
+		if err := db.PutUpstreamHold("registry.npmjs.org", now.Add(time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+		// A shorter hold reported later keeps the longer one.
+		if err := db.PutUpstreamHold("registry.npmjs.org", now.Add(30*time.Second)); err != nil {
+			t.Fatal(err)
+		}
+		if err := db.PutUpstreamHold("pypi.org", now.Add(10*time.Second)); err != nil {
+			t.Fatal(err)
+		}
+		holds, err := db.UpstreamHolds(now)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(holds) != 2 || !holds["registry.npmjs.org"].Equal(now.Add(time.Minute)) {
+			t.Fatalf("holds = %v", holds)
+		}
+
+		// A longer one replaces it.
+		if err := db.PutUpstreamHold("registry.npmjs.org", now.Add(time.Hour)); err != nil {
+			t.Fatal(err)
+		}
+		holds, err = db.UpstreamHolds(now.Add(20 * time.Second))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(holds) != 1 || !holds["registry.npmjs.org"].Equal(now.Add(time.Hour)) {
+			t.Fatalf("holds after pypi.org ended = %v", holds)
+		}
+		var rows int
+		if err := db.Get(&rows, "SELECT COUNT(*) FROM upstream_holds"); err != nil {
+			t.Fatal(err)
+		}
+		if rows != 1 {
+			t.Errorf("rows = %d, want the ended hold deleted", rows)
+		}
+	})
+}

internal/handler/handler.go

@@ -26,7 +26,6 @@ import (
 	"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"
 )
@@ -195,7 +194,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 *registryclient.RateLimitGate
+	RateLimits UpstreamHolds
 
 	// HostedNPMScopes and HostedComposerVendors name namespaces served from
 	// pushed packages alone, without asking the upstream.
@@ -584,7 +583,7 @@ func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, fil
 		}
 	}
 
-	if err := p.RateLimits.Held(info.URL); err != nil {
+	if err := p.upstreamHeld(info.URL); err != nil {
 		return artifacts.Artifact{}, "", err
 	}
 	p.Logger.Info("fetching from upstream",
@@ -874,7 +873,12 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch
 	}
 	// With leaseFailed, release is a no-op and this instance fetches on its
 	// own, so a failing upstream costs each instance one attempt in
-	// parallel.
+	// parallel. A rate-limited holder has published its hold by now, and
+	// rereading it makes commit refuse the fetch with the holder's
+	// Retry-After instead of sending one more request to the limited host.
+	if outcome == leaseFailed {
+		p.reloadUpstreamHolds()
+	}
 	failed := true
 	defer func() { release(failed) }()
 
@@ -1441,7 +1445,10 @@ func (p *Proxy) coalescedMetadataMiss(ctx context.Context, ecosystem, cacheKey,
 				return *hit
 			case leaseFailed:
 				// Serve the stale copy at once, as the holder did, rather
-				// than trying the failing upstream again. Without one, fetch.
+				// than trying the failing upstream again. Without one, fetch,
+				// unless the holder was rate limited: its hold, read here,
+				// refuses the fetch with its Retry-After.
+				p.reloadUpstreamHolds()
 				if entry != nil {
 					if res := p.staleMetadata(fetchCtx, ecosystem, cacheKey, validate, entry, errFetchFailedElsewhere); res.err == nil {
 						return res
@@ -2202,7 +2209,7 @@ 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 {
+	if err := p.upstreamHeld(downloadURL); err != nil {
 		return artifacts.Artifact{}, "", err
 	}
 	p.Logger.Info("fetching from upstream",
@@ -2269,7 +2276,7 @@ func (p *Proxy) fetchUpstream(ctx context.Context, downloadURL string, headers h
 // 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 {
+	if err := p.upstreamHeld(upstreamURL); err != nil {
 		return nil, err
 	}
 	p.Logger.Info("streaming from upstream",