From cd7c2c43837825e1507374a85bd2652947164d78 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Thu, 13 Aug 2026 17:23:59 +1000 Subject: [PATCH] remotes: add least-connections mirror strategy (round-robin remains default) (#122) ## Why The mirrorlist (PR #121) always load-balances round-robin. Round-robin is oblivious to how busy each mirror is, so a slow or saturated mirror keeps getting its fair share of new requests. This adds an opt-in **least-connections** strategy that favors the mirror currently handling the fewest in-flight requests, steering new work toward idle mirrors. Round-robin stays the default, so existing remotes are unchanged. ## How - **Model**: `models.Remote` gains `mirror_strategy` (`round_robin` default/empty, or `least_conn`). `ValidateMirrorStrategy` checks the enum and requires a non-empty mirrorlist for `least_conn`. Empty behaves as `round_robin` for back-compat. - **DB**: additive `mirror_strategy TEXT NOT NULL DEFAULT 'round_robin'` column (CREATE TABLE + `ADD COLUMN IF NOT EXISTS`), wired through remoteCols/scanRemote/CreateRemote/UpdateRemote; empty normalized to `round_robin` on write. - **Engine**: a per-remote, per-pool-URL atomic in-flight gauge is incremented around each upstream call (head/fetch/checkUpstream). For a `least_conn` remote the attempt order starts with the least-loaded pool URL (ties broken by the existing round-robin rotation). Round-robin path and failover order are unchanged; single-URL pools are a no-op. - **Tests**: unit tests for least-loaded selection, round-robin default, gauge inc/dec, single-URL no-op, and validation; DB round-trip covers the new column; docker e2e adds a `least_conn` distribution test plus a real `dnf` install through a `least_conn` remote. Back-compat: unset/empty `mirror_strategy` is `round_robin`, so all existing remotes keep their current behavior. Reviewed-on: https://git.unkin.net/unkin/artifactapi/pulls/122 Co-authored-by: unkin-agent Co-committed-by: unkin-agent --- e2e-docker/README.md | 6 ++ e2e-docker/leastconn_test.go | 105 +++++++++++++++++++++++++++++ internal/api/v2/remotes.go | 8 +++ internal/database/database_test.go | 14 ++++ internal/database/postgres.go | 2 + internal/database/remotes.go | 21 ++++-- internal/proxy/engine.go | 54 ++++++++++++++- internal/proxy/leastconn_test.go | 103 ++++++++++++++++++++++++++++ pkg/models/remote.go | 35 ++++++++-- pkg/models/remote_test.go | 24 +++++++ 10 files changed, 362 insertions(+), 10 deletions(-) create mode 100644 e2e-docker/leastconn_test.go create mode 100644 internal/proxy/leastconn_test.go diff --git a/e2e-docker/README.md b/e2e-docker/README.md index f6ab089..d57bf79 100644 --- a/e2e-docker/README.md +++ b/e2e-docker/README.md @@ -36,6 +36,12 @@ already-running stack. no-mirrorlist regression, and a real `dnf` (stock `rockylinux:9` container) `makecache` + `install` through a two-mirror rpm remote whose `base_url` is dead — a dead mirror must not break the client. +- **Mirror strategy (`least_conn`)** — a `mirror_strategy: least_conn` rpm remote + over the two constant-body mirrors exercises the least-connections selection + path end-to-end (both mirrors serve, all requests succeed), plus a real `dnf` + install through a `least_conn` remote with a dead primary (failover unchanged). + The precise least-loaded pick is asserted deterministically in the proxy unit + test, since an in-flight-skew assertion over HTTP is timing-sensitive. ## Fixtures diff --git a/e2e-docker/leastconn_test.go b/e2e-docker/leastconn_test.go new file mode 100644 index 0000000..43a49b0 --- /dev/null +++ b/e2e-docker/leastconn_test.go @@ -0,0 +1,105 @@ +//go:build dockere2e + +package e2edocker + +import ( + "fmt" + "net/http" + "os" + "os/exec" + "strings" + "testing" +) + +// TestLeastConnMultiBaseURL configures an rpm remote with mirror_strategy = +// least_conn over a two-mirror pool and drives distinct cache-miss paths through +// it, asserting every request succeeds and both mirrors serve traffic. This +// exercises the least-connections selection path (leastConnOrder + the in-flight +// gauge inc/dec around each upstream call) end-to-end through a real HTTP client. +// A precise least-loaded assertion is timing-sensitive over HTTP and is covered +// deterministically by the proxy unit test (TestLeastConnPicksLeastLoaded); +// here, with requests issued serially, in-flight counts return to zero between +// them so equal-load mirrors are spread by the round-robin tie-break. +func TestLeastConnMultiBaseURL(t *testing.T) { + name := "e2e-leastconn" + createRepo(t, fmt.Sprintf(`{ + "name": %q, + "package_type": "rpm", + "repo_type": "remote", + "base_url": %q, + "mirrorlist": [%q], + "mirror_strategy": "least_conn", + "stale_on_error": false + }`, name, mockUpstreamA(), mockUpstreamB())) + defer deleteRepo(t, name) + + seenA, seenB := false, false + const n = 12 + for i := 0; i < n; i++ { + url := api(fmt.Sprintf("/api/v1/remote/%s/lc/%d", name, i)) + resp, body := doRequest(t, http.MethodGet, url, nil, "") + if resp.StatusCode != http.StatusOK { + t.Fatalf("request %d: status %d: %s", i, resp.StatusCode, body) + } + switch strings.TrimSpace(string(body)) { + case "UPSTREAM-A": + seenA = true + case "UPSTREAM-B": + seenB = true + default: + t.Fatalf("request %d: unexpected body %q", i, body) + } + } + if !seenA || !seenB { + t.Fatalf("least_conn remote did not reach both upstreams: A=%v B=%v", seenA, seenB) + } +} + +// TestLeastConnDnfInstall drives a real dnf (stock rockylinux container) at a +// two-mirror rpm remote configured with mirror_strategy = least_conn whose +// primary base_url is dead: makecache + install must succeed via the live mirror. +// This proves a real package-manager client installs correctly through a +// least_conn remote and that failover semantics are unchanged under the new +// strategy. Requires the compose network exported by scripts/docker-e2e.sh. +func TestLeastConnDnfInstall(t *testing.T) { + network := os.Getenv("COMPOSE_NETWORK") + internal := os.Getenv("ARTIFACTAPI_INTERNAL") + if network == "" || internal == "" { + t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; run via scripts/docker-e2e.sh") + } + if _, err := exec.LookPath("docker"); err != nil { + t.Skip("docker not available on the test host") + } + + name := "e2e-leastconn-dnf" + createRepo(t, fmt.Sprintf(`{ + "name": %q, + "package_type": "rpm", + "repo_type": "remote", + "base_url": "http://mockupstream-dead:80", + "mirrorlist": [%q], + "mirror_strategy": "least_conn", + "stale_on_error": false + }`, name, mockUpstream())) + defer deleteRepo(t, name) + + repoURL := strings.TrimRight(internal, "/") + "/api/v1/remote/" + name + "/rpm-mirror" + repoConf := fmt.Sprintf("[dnflc]\nname=dnflc\nbaseurl=%s\nenabled=1\ngpgcheck=0\nsslverify=0\nmetadata_expire=0\n", repoURL) + script := "set -euo pipefail; " + + "printf '%s' \"$REPO\" > /etc/yum.repos.d/dnflc.repo; " + + "dnf -y --disablerepo='*' --enablerepo=dnflc makecache; " + + "dnf -y --disablerepo='*' --enablerepo=dnflc install e2e-testpkg; " + + "rpm -q e2e-testpkg" + + cmd := exec.Command("docker", "run", "--rm", + "--network", network, + "-e", "REPO="+repoConf, + "rockylinux:9", "bash", "-c", script) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("real dnf install through a least_conn remote failed: %v\n%s", err, out) + } + if !strings.Contains(string(out), "e2e-testpkg-1.0-1") { + t.Fatalf("dnf did not install the expected package via least_conn remote; output:\n%s", out) + } +} diff --git a/internal/api/v2/remotes.go b/internal/api/v2/remotes.go index 5e3047d..a8f679e 100644 --- a/internal/api/v2/remotes.go +++ b/internal/api/v2/remotes.go @@ -92,6 +92,10 @@ func (h *RemotesHandler) create(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } + if err := remote.ValidateMirrorStrategy(); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } if err := remote.ValidatePatterns(); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -120,6 +124,10 @@ func (h *RemotesHandler) update(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } + if err := remote.ValidateMirrorStrategy(); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } if err := remote.ValidatePatterns(); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return diff --git a/internal/database/database_test.go b/internal/database/database_test.go index fa72d54..ad1fcf7 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -120,6 +120,20 @@ func TestRemoteMirrorlistRoundTrip(t *testing.T) { if len(got.Mirrorlist) != 2 || got.Mirrorlist[0] != mirrors[0] || got.Mirrorlist[1] != mirrors[1] { t.Fatalf("Mirrorlist round-trip = %v, want %v", got.Mirrorlist, mirrors) } + // An unset strategy is stored as the round_robin default. + if got.MirrorStrategy != models.MirrorStrategyRoundRobin { + t.Fatalf("MirrorStrategy default = %q, want %q", got.MirrorStrategy, models.MirrorStrategyRoundRobin) + } + + // Updating to least_conn round-trips. + got.MirrorStrategy = models.MirrorStrategyLeastConn + if err := testDB.UpdateRemote(ctx(), got); err != nil { + t.Fatalf("update to least_conn: %v", err) + } + got, _ = testDB.GetRemote(ctx(), "r-mirror") + if got.MirrorStrategy != models.MirrorStrategyLeastConn { + t.Fatalf("MirrorStrategy after update = %q, want least_conn", got.MirrorStrategy) + } // Clearing the mirrorlist on update persists an empty list. got.Mirrorlist = nil diff --git a/internal/database/postgres.go b/internal/database/postgres.go index d4aa3c8..4e5839a 100644 --- a/internal/database/postgres.go +++ b/internal/database/postgres.go @@ -45,6 +45,7 @@ func (db *DB) migrate() error { repo_type TEXT DEFAULT 'remote', base_url TEXT NOT NULL DEFAULT '', mirrorlist TEXT[] DEFAULT '{}', + mirror_strategy TEXT NOT NULL DEFAULT 'round_robin', description TEXT DEFAULT '', username TEXT DEFAULT '', password TEXT DEFAULT '', @@ -126,6 +127,7 @@ func (db *DB) migrate() error { ALTER TABLE remotes ADD COLUMN IF NOT EXISTS repo_type TEXT DEFAULT 'remote'; ALTER TABLE remotes ADD COLUMN IF NOT EXISTS mirrorlist TEXT[] DEFAULT '{}'; + ALTER TABLE remotes ADD COLUMN IF NOT EXISTS mirror_strategy TEXT NOT NULL DEFAULT 'round_robin'; ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_dial_timeout INTEGER DEFAULT 0; ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_tls_timeout INTEGER DEFAULT 0; ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_response_header_timeout INTEGER DEFAULT 0; diff --git a/internal/database/remotes.go b/internal/database/remotes.go index 5eadfda..76afa06 100644 --- a/internal/database/remotes.go +++ b/internal/database/remotes.go @@ -6,7 +6,7 @@ import ( "git.unkin.net/unkin/artifactapi/pkg/models" ) -const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, description, username, password, +const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, mirror_strategy, description, username, password, immutable_ttl, mutable_ttl, check_mutable, patterns, blocklist, mutable_patterns, immutable_patterns, ban_tags_enabled, ban_tags, @@ -15,9 +15,18 @@ const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, descrip upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout, created_at, updated_at` +// normalizeMirrorStrategy maps an empty strategy to the round_robin default so +// the NOT NULL mirror_strategy column always stores a canonical value. +func normalizeMirrorStrategy(s string) string { + if s == "" { + return models.MirrorStrategyRoundRobin + } + return s +} + func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error { return scanner.Scan( - &r.Name, &r.PackageType, &r.RepoType, &r.BaseURL, &r.Mirrorlist, &r.Description, &r.Username, &r.Password, + &r.Name, &r.PackageType, &r.RepoType, &r.BaseURL, &r.Mirrorlist, &r.MirrorStrategy, &r.Description, &r.Username, &r.Password, &r.ImmutableTTL, &r.MutableTTL, &r.CheckMutable, &r.Patterns, &r.Blocklist, &r.MutablePatterns, &r.ImmutablePatterns, &r.BanTagsEnabled, &r.BanTags, @@ -64,8 +73,9 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error { ban_tags_enabled, ban_tags, quarantine_enabled, quarantine_days, stale_on_error, releases_remote, managed_by, - upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25) + upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout, + mirror_strategy + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26) `, r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Mirrorlist, r.Description, r.Username, r.Password, r.ImmutableTTL, r.MutableTTL, r.CheckMutable, @@ -74,6 +84,7 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error { r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError, r.ReleasesRemote, r.ManagedBy, r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout, + normalizeMirrorStrategy(r.MirrorStrategy), ) return err } @@ -88,6 +99,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error { quarantine_enabled=$17, quarantine_days=$18, stale_on_error=$19, releases_remote=$20, managed_by=$21, upstream_dial_timeout=$22, upstream_tls_timeout=$23, upstream_response_header_timeout=$24, + mirror_strategy=$26, updated_at=NOW() WHERE name=$1 `, @@ -99,6 +111,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error { r.ReleasesRemote, r.ManagedBy, r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout, r.Mirrorlist, + normalizeMirrorStrategy(r.MirrorStrategy), ) return err } diff --git a/internal/proxy/engine.go b/internal/proxy/engine.go index b7b89bb..c39ca98 100644 --- a/internal/proxy/engine.go +++ b/internal/proxy/engine.go @@ -10,6 +10,7 @@ import ( "io" "log/slog" "net/http" + "sort" "strings" "sync" "sync/atomic" @@ -41,6 +42,11 @@ type Engine struct { // *atomic.Uint64) used to rotate the starting mirror across upstream base // URLs. Distribution is per-replica and approximate, which is fine. rrCounters sync.Map + // inflight holds a per-remote, per-upstream-URL in-flight request gauge + // (key "remoteName\x00baseURL" -> *atomic.Int64) used by the least_conn + // mirror strategy to prefer the mirror currently handling the fewest + // requests. Per-replica and approximate, which is fine. + inflight sync.Map } func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine { @@ -237,7 +243,9 @@ func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path st } var lastErr error for i, url := range order { + ctr := e.beginAttempt(remote, url) result, err := e.headUpstreamOnce(ctx, withBaseURL(remote, url), path, prov) + endAttempt(ctr) if err == nil { return result, nil } @@ -316,7 +324,9 @@ func (e *Engine) fetchFromUpstream(ctx context.Context, remote models.Remote, pa } var lastErr error for i, url := range order { + ctr := e.beginAttempt(remote, url) result, err := e.fetchFromUpstreamOnce(ctx, withBaseURL(remote, url), path, prov, class, ttl, clientHeaders) + endAttempt(ctr) if err == nil { return result, nil } @@ -517,7 +527,9 @@ func (e *Engine) checkUpstream(ctx context.Context, remote models.Remote, path, } var lastErr error for i, url := range order { + ctr := e.beginAttempt(remote, url) notModified, err := e.checkUpstreamOnce(ctx, withBaseURL(remote, url), path, etag, prov) + endAttempt(ctr) if err == nil { return notModified, nil } @@ -729,23 +741,61 @@ func isNetworkError(err error) bool { // baseURLAttemptOrder returns the ordered upstream base URLs to try for a single // request, drawn from the remote's pool ([base_url] + mirrorlist). A multi-mirror -// remote starts at the next round-robin position and advances linearly for +// remote starts at a strategy-chosen position and advances linearly for // failover; a remote with no mirrorlist yields exactly [base_url], preserving the -// original single-attempt behavior. +// original single-attempt behavior. The default (round_robin) rotates the +// starting mirror; least_conn starts with the mirror handling the fewest +// in-flight requests. Failover order after the first pick is unchanged. func (e *Engine) baseURLAttemptOrder(remote models.Remote) []string { urls := remote.UpstreamPool() if len(urls) <= 1 { return urls } + // Rotate by the round-robin cursor first so equal-load mirrors still spread + // evenly; least_conn then stable-sorts this rotation by in-flight count. v, _ := e.rrCounters.LoadOrStore(remote.Name, new(atomic.Uint64)) start := int(v.(*atomic.Uint64).Add(1) - 1) ordered := make([]string, len(urls)) for i := range urls { ordered[i] = urls[(start+i)%len(urls)] } + if remote.MirrorStrategy == models.MirrorStrategyLeastConn { + sort.SliceStable(ordered, func(a, b int) bool { + return e.inflightCounter(remote.Name, ordered[a]).Load() < e.inflightCounter(remote.Name, ordered[b]).Load() + }) + } return ordered } +// inflightCounter returns the shared in-flight request gauge for a given +// (remote, upstream URL), creating it on first use. +func (e *Engine) inflightCounter(remoteName, url string) *atomic.Int64 { + v, _ := e.inflight.LoadOrStore(remoteName+"\x00"+url, new(atomic.Int64)) + return v.(*atomic.Int64) +} + +// beginAttempt increments the in-flight gauge for a least_conn multi-mirror +// remote before an upstream call and returns the counter to release; it is a +// no-op (returns nil) for round-robin remotes and single-URL pools. +func (e *Engine) beginAttempt(remote models.Remote, url string) *atomic.Int64 { + if remote.MirrorStrategy != models.MirrorStrategyLeastConn { + return nil + } + if len(remote.UpstreamPool()) <= 1 { + return nil + } + ctr := e.inflightCounter(remote.Name, url) + ctr.Add(1) + return ctr +} + +// endAttempt decrements a gauge returned by beginAttempt, tolerating nil. +func endAttempt(ctr *atomic.Int64) { + if ctr != nil { + ctr.Add(-1) + } +} + // withBaseURL narrows a remote's active BaseURL to a single selected mirror so // providers (UpstreamURL/AuthHeaders/RewriteResponse) operate on exactly that // upstream for this attempt. diff --git a/internal/proxy/leastconn_test.go b/internal/proxy/leastconn_test.go new file mode 100644 index 0000000..097f921 --- /dev/null +++ b/internal/proxy/leastconn_test.go @@ -0,0 +1,103 @@ +package proxy + +import ( + "testing" + + "git.unkin.net/unkin/artifactapi/pkg/models" +) + +// baseURLAttemptOrder and the in-flight gauge only touch the engine's sync.Map +// fields, so these tests run against a zero-value Engine without a DB/S3/redis +// stack and can drive the gauge deterministically. + +// TestLeastConnPicksLeastLoaded pre-loads one mirror's in-flight gauge and +// asserts a least_conn remote starts its attempt order with the idle mirror. +func TestLeastConnPicksLeastLoaded(t *testing.T) { + e := &Engine{} + r := models.Remote{ + Name: "lc", + BaseURL: "https://a.example", + Mirrorlist: []string{"https://b.example"}, + MirrorStrategy: models.MirrorStrategyLeastConn, + } + + // Make A appear busy: least_conn must prefer B regardless of RR rotation. + e.inflightCounter(r.Name, "https://a.example").Add(3) + for i := 0; i < 5; i++ { + order := e.baseURLAttemptOrder(r) + if len(order) != 2 { + t.Fatalf("attempt %d: order len = %d, want 2", i, len(order)) + } + if order[0] != "https://b.example" { + t.Fatalf("attempt %d: least_conn started with %q, want idle mirror https://b.example", i, order[0]) + } + } + + // Once B is the busier mirror, the starting pick flips to A. + e.inflightCounter(r.Name, "https://b.example").Add(10) + if order := e.baseURLAttemptOrder(r); order[0] != "https://a.example" { + t.Fatalf("after loading B, least_conn started with %q, want https://a.example", order[0]) + } +} + +// TestRoundRobinDefaultUnchanged asserts an unset strategy still rotates the +// starting mirror across the pool and ignores the in-flight gauge. +func TestRoundRobinDefaultUnchanged(t *testing.T) { + e := &Engine{} + r := models.Remote{ + Name: "rr", + BaseURL: "https://a.example", + Mirrorlist: []string{"https://b.example"}, + } + + // Even with A heavily loaded, round-robin must still rotate (not avoid A). + e.inflightCounter(r.Name, "https://a.example").Add(100) + starts := map[string]int{} + for i := 0; i < 4; i++ { + starts[e.baseURLAttemptOrder(r)[0]]++ + } + if starts["https://a.example"] == 0 || starts["https://b.example"] == 0 { + t.Fatalf("round-robin did not rotate starting mirror: %v", starts) + } +} + +// TestLeastConnSingleURLNoOp asserts a single-URL pool yields exactly [base_url] +// and beginAttempt is a no-op there and for round-robin remotes. +func TestLeastConnSingleURLNoOp(t *testing.T) { + e := &Engine{} + solo := models.Remote{Name: "solo", BaseURL: "https://a.example", MirrorStrategy: models.MirrorStrategyLeastConn} + if order := e.baseURLAttemptOrder(solo); len(order) != 1 || order[0] != "https://a.example" { + t.Fatalf("single-url order = %v, want [base_url]", order) + } + if ctr := e.beginAttempt(solo, "https://a.example"); ctr != nil { + t.Fatal("beginAttempt on single-url pool should be a no-op (nil)") + } + + rr := models.Remote{Name: "rr2", BaseURL: "https://a.example", Mirrorlist: []string{"https://b.example"}} + if ctr := e.beginAttempt(rr, "https://a.example"); ctr != nil { + t.Fatal("beginAttempt on round-robin remote should be a no-op (nil)") + } +} + +// TestBeginEndAttemptGauge asserts the gauge increments on begin and returns to +// zero after endAttempt, so it tracks live in-flight requests. +func TestBeginEndAttemptGauge(t *testing.T) { + e := &Engine{} + r := models.Remote{ + Name: "g", + BaseURL: "https://a.example", + Mirrorlist: []string{"https://b.example"}, + MirrorStrategy: models.MirrorStrategyLeastConn, + } + c1 := e.beginAttempt(r, "https://a.example") + c2 := e.beginAttempt(r, "https://a.example") + if got := e.inflightCounter(r.Name, "https://a.example").Load(); got != 2 { + t.Fatalf("gauge after two begins = %d, want 2", got) + } + endAttempt(c1) + endAttempt(c2) + if got := e.inflightCounter(r.Name, "https://a.example").Load(); got != 0 { + t.Fatalf("gauge after matching ends = %d, want 0", got) + } + endAttempt(nil) // tolerated +} diff --git a/pkg/models/remote.go b/pkg/models/remote.go index 2b50891..fd27aa6 100644 --- a/pkg/models/remote.go +++ b/pkg/models/remote.go @@ -43,10 +43,14 @@ type Remote struct { // Mirrorlist holds additional upstream mirror base URLs. The effective // upstream pool is [base_url] + mirrorlist, load-balanced round-robin with // failover by the proxy engine. Only valid on remote rpm/deb/apk repos. - Mirrorlist []string `json:"mirrorlist,omitempty"` - Description string `json:"description,omitempty"` - Username string `json:"-"` - Password string `json:"-"` + Mirrorlist []string `json:"mirrorlist,omitempty"` + // MirrorStrategy selects how the proxy engine picks the starting upstream + // from the pool: round_robin (default/empty) rotates, least_conn favors the + // mirror with the fewest in-flight requests. Failover order is unchanged. + MirrorStrategy string `json:"mirror_strategy,omitempty"` + Description string `json:"description,omitempty"` + Username string `json:"-"` + Password string `json:"-"` ImmutableTTL int `json:"immutable_ttl"` MutableTTL int `json:"mutable_ttl"` @@ -77,6 +81,13 @@ type Remote struct { UpdatedAt time.Time `json:"updated_at"` } +// Mirror balancing strategies for MirrorStrategy. An empty value is treated as +// round_robin, so existing remotes keep their current behavior. +const ( + MirrorStrategyRoundRobin = "round_robin" + MirrorStrategyLeastConn = "least_conn" +) + // mirrorlistPackageTypes are the package types for which a mirrorlist is // allowed: OS package repos (rpm, deb, apk/alpine) that fetch many small files // and benefit most from mirror load-balancing and failover. @@ -122,6 +133,22 @@ func (r *Remote) ValidateMirrorlist() error { return nil } +// ValidateMirrorStrategy enforces that mirror_strategy is one of the allowed +// values and that a non-default strategy (least_conn) is only set alongside a +// non-empty mirrorlist, where balancing is meaningful. An empty strategy is +// accepted and behaves as round_robin. +func (r *Remote) ValidateMirrorStrategy() error { + switch r.MirrorStrategy { + case "", MirrorStrategyRoundRobin, MirrorStrategyLeastConn: + default: + return fmt.Errorf("invalid mirror_strategy %q: must be %q or %q", r.MirrorStrategy, MirrorStrategyRoundRobin, MirrorStrategyLeastConn) + } + if r.MirrorStrategy == MirrorStrategyLeastConn && len(r.Mirrorlist) == 0 { + return fmt.Errorf("mirror_strategy %q requires a non-empty mirrorlist", r.MirrorStrategy) + } + return nil +} + // ValidatePatterns ensures every configured regex compiles. Storing an // invalid pattern would otherwise be silently dropped at match time, which // for the blocklist is a fail-open: a mistyped deny rule becomes a no-op. diff --git a/pkg/models/remote_test.go b/pkg/models/remote_test.go index c5749ac..7f757f4 100644 --- a/pkg/models/remote_test.go +++ b/pkg/models/remote_test.go @@ -74,6 +74,30 @@ func TestUpstreamPool(t *testing.T) { } } +func TestValidateMirrorStrategy(t *testing.T) { + ml := []string{"https://m.example"} + cases := []struct { + name string + remote Remote + wantErr bool + }{ + {"empty defaults ok", Remote{Mirrorlist: ml}, false}, + {"explicit round_robin ok", Remote{MirrorStrategy: MirrorStrategyRoundRobin, Mirrorlist: ml}, false}, + {"round_robin without mirrorlist ok", Remote{MirrorStrategy: MirrorStrategyRoundRobin}, false}, + {"least_conn with mirrorlist ok", Remote{MirrorStrategy: MirrorStrategyLeastConn, Mirrorlist: ml}, false}, + {"least_conn without mirrorlist rejected", Remote{MirrorStrategy: MirrorStrategyLeastConn}, true}, + {"unknown strategy rejected", Remote{MirrorStrategy: "random", Mirrorlist: ml}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := tc.remote.ValidateMirrorStrategy() + if (err != nil) != tc.wantErr { + t.Errorf("ValidateMirrorStrategy() err = %v, wantErr = %v", err, tc.wantErr) + } + }) + } +} + func TestValidateMirrorlist(t *testing.T) { cases := []struct { name string