remotes: add mirrorlist for round-robin + failover across mirrors (rpm/deb/apk)

OS package remotes (rpm/deb/apk) fetch many small files and benefit from spreading upstream load across mirrors and surviving a mirror outage. A remote may now set a `mirrorlist` of additional upstream base URLs; the effective upstream pool is [base_url] + mirrorlist, which the shared proxy engine load-balances round-robin and, on a network error/timeout/5xx, fails over to the next mirror before returning an error. Because selection happens in the engine, it works for every provider that reaches upstream. Backward compatible: `base_url` stays a plain string that providers read unchanged, and a remote with no mirrorlist behaves exactly as today (single attempt, same error path). - add models.Remote.Mirrorlist ([]string, json "mirrorlist,omitempty") and UpstreamPool() = [base_url] + mirrorlist; ValidateMirrorlist enforces remote repo_type + package_type in {rpm, deb, alpine} and http/https URLs - v2 create/update: reject a mirrorlist on any other repo (400); base_url remains required for remotes - persist mirrorlist in a new additive `mirrorlist TEXT[]` column (remoteCols/scanRemote/CreateRemote/UpdateRemote); base_url column unchanged - engine: per-remote round-robin cursor over the pool; wrap the fetch/head/revalidate upstream calls in a failover loop that narrows the remote to one selected mirror per attempt; only network errors and 5xx fail over (404/403/... return as-is); circuit breaker stays keyed per remote and trips only after all mirrors fail - tests: model JSON round-trip + validation gating, engine round-robin/failover/no-mirrorlist-unchanged, DB mirrorlist round-trip, and a docker acceptance suite (round-robin across two mock upstreams, failover past a dead primary, no-mirrorlist regression, and a real dnf makecache+install through a two-mirror rpm remote whose base_url is dead) Least-connections and a per-remote strategy selector are a follow-up PR.
2026-08-13 08:26:17 +10:00
parent 73c0bfc670
commit ac8be0a105
22 changed files with 763 additions and 13 deletions
@@ -11,6 +11,8 @@ import (
 	"log/slog"
 	"net/http"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"

 	"git.unkin.net/unkin/artifactapi/internal/cache"
@@ -35,6 +37,10 @@ type Engine struct {
 	cas       *storage.CAS
 	circuit   *CircuitBreaker
 	accessLog chan database.AccessLogEntry
+	// rrCounters holds a per-remote round-robin cursor (remoteName ->
+	// *atomic.Uint64) used to rotate the starting mirror across upstream base
+	// URLs. Distribution is per-replica and approximate, which is fine.
+	rrCounters sync.Map
 }

 func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine {
@@ -222,7 +228,30 @@ func (e *Engine) Head(ctx context.Context, remote models.Remote, path string, pr
 	return e.headUpstream(ctx, remote, path, prov)
 }

+// headUpstream issues an upstream HEAD, load-balancing across the remote's base
+// URLs and failing over to the next mirror on a network error or 5xx.
 func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path string, prov provider.Provider) (*HeadResult, error) {
+	order := e.baseURLAttemptOrder(remote)
+	if len(order) == 0 {
+		return nil, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
+	}
+	var lastErr error
+	for i, url := range order {
+		result, err := e.headUpstreamOnce(ctx, withBaseURL(remote, url), path, prov)
+		if err == nil {
+			return result, nil
+		}
+		lastErr = err
+		if i < len(order)-1 && shouldFailover(err) {
+			slog.Warn("upstream HEAD failed, failing over", "remote", remote.Name, "base_url", url, "error", err)
+			continue
+		}
+		return nil, err
+	}
+	return nil, lastErr
+}
+
+func (e *Engine) headUpstreamOnce(ctx context.Context, remote models.Remote, path string, prov provider.Provider) (*HeadResult, error) {
 	url := prov.UpstreamURL(remote, path)

 	authHeaders, err := prov.AuthHeaders(ctx, remote)
@@ -277,7 +306,31 @@ func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path st
 	return &HeadResult{ContentType: contentType, Size: resp.ContentLength, Source: "remote"}, nil
 }

+// fetchFromUpstream fetches an artifact from upstream, load-balancing across the
+// remote's base URLs and failing over to the next mirror on a network error or
+// 5xx before returning an error.
 func (e *Engine) fetchFromUpstream(ctx context.Context, remote models.Remote, path string, prov provider.Provider, class Classification, ttl time.Duration, clientHeaders http.Header) (*FetchResult, error) {
+	order := e.baseURLAttemptOrder(remote)
+	if len(order) == 0 {
+		return nil, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
+	}
+	var lastErr error
+	for i, url := range order {
+		result, err := e.fetchFromUpstreamOnce(ctx, withBaseURL(remote, url), path, prov, class, ttl, clientHeaders)
+		if err == nil {
+			return result, nil
+		}
+		lastErr = err
+		if i < len(order)-1 && shouldFailover(err) {
+			slog.Warn("upstream fetch failed, failing over", "remote", remote.Name, "base_url", url, "error", err)
+			continue
+		}
+		return nil, err
+	}
+	return nil, lastErr
+}
+
+func (e *Engine) fetchFromUpstreamOnce(ctx context.Context, remote models.Remote, path string, prov provider.Provider, class Classification, ttl time.Duration, clientHeaders http.Header) (*FetchResult, error) {
 	url := prov.UpstreamURL(remote, path)

 	authHeaders, err := prov.AuthHeaders(ctx, remote)
@@ -454,7 +507,31 @@ func (e *Engine) serveFromStore(ctx context.Context, remote models.Remote, path
 	}, nil
 }

+// checkUpstream issues a conditional upstream HEAD (If-None-Match), load
+// balancing across the remote's base URLs and failing over to the next mirror on
+// a network error or 5xx.
 func (e *Engine) checkUpstream(ctx context.Context, remote models.Remote, path, etag string, prov provider.Provider) (bool, error) {
+	order := e.baseURLAttemptOrder(remote)
+	if len(order) == 0 {
+		return false, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
+	}
+	var lastErr error
+	for i, url := range order {
+		notModified, err := e.checkUpstreamOnce(ctx, withBaseURL(remote, url), path, etag, prov)
+		if err == nil {
+			return notModified, nil
+		}
+		lastErr = err
+		if i < len(order)-1 && shouldFailover(err) {
+			slog.Warn("upstream revalidation failed, failing over", "remote", remote.Name, "base_url", url, "error", err)
+			continue
+		}
+		return false, err
+	}
+	return false, lastErr
+}
+
+func (e *Engine) checkUpstreamOnce(ctx context.Context, remote models.Remote, path, etag string, prov provider.Provider) (bool, error) {
 	url := prov.UpstreamURL(remote, path)

 	req, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil)
@@ -649,3 +726,45 @@ func isNetworkError(err error) bool {
 	var ue *UpstreamError
 	return errors.As(err, &ue)
 }
+
+// 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
+// failover; a remote with no mirrorlist yields exactly [base_url], preserving the
+// original single-attempt behavior.
+func (e *Engine) baseURLAttemptOrder(remote models.Remote) []string {
+	urls := remote.UpstreamPool()
+	if len(urls) <= 1 {
+		return urls
+	}
+	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)]
+	}
+	return ordered
+}
+
+// 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.
+func withBaseURL(remote models.Remote, url string) models.Remote {
+	remote.BaseURL = url
+	remote.Mirrorlist = nil
+	return remote
+}
+
+// shouldFailover reports whether an upstream attempt error is worth retrying
+// against the next mirror: network errors/timeouts and upstream 5xx responses.
+// Definitive statuses (404/403/401/...) are returned to the caller unchanged.
+func shouldFailover(err error) bool {
+	if isNetworkError(err) {
+		return true
+	}
+	var pe *ProxyError
+	if errors.As(err, &pe) {
+		return pe.Status >= 500
+	}
+	return false
+}
@@ -0,0 +1,188 @@
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+// TestFetchMultiBaseURLRoundRobin drives distinct artifact paths through a
+// remote configured with two upstreams and asserts both receive traffic.
+func TestFetchMultiBaseURLRoundRobin(t *testing.T) {
+	requireStack(t)
+	ctx := context.Background()
+
+	var hitsA, hitsB atomic.Int64
+	upA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hitsA.Add(1)
+		w.Write([]byte("A"))
+	}))
+	defer upA.Close()
+	upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hitsB.Add(1)
+		w.Write([]byte("B"))
+	}))
+	defer upB.Close()
+
+	r := seed(t, models.Remote{
+		Name:         "eng-rr",
+		PackageType:  models.PackageGeneric,
+		RepoType:     models.RepoTypeRemote,
+		BaseURL:      upA.URL,
+		Mirrorlist:   []string{upB.URL},
+		StaleOnError: true,
+	})
+	p := prov(t, models.PackageGeneric)
+
+	const n = 10
+	for i := 0; i < n; i++ {
+		res, err := testEngine.Fetch(ctx, r, fmt.Sprintf("rr-%d.bin", i), p)
+		if err != nil {
+			t.Fatalf("fetch %d: %v", i, err)
+		}
+		res.Reader.Close()
+	}
+
+	if hitsA.Load() == 0 || hitsB.Load() == 0 {
+		t.Fatalf("round-robin did not spread across both upstreams: A=%d B=%d", hitsA.Load(), hitsB.Load())
+	}
+	if total := hitsA.Load() + hitsB.Load(); total != n {
+		t.Fatalf("expected %d upstream hits total, got %d (A=%d B=%d)", n, total, hitsA.Load(), hitsB.Load())
+	}
+}
+
+// TestFetchMultiBaseURLFailover asserts that a dead/erroring primary mirror
+// transparently fails over to a healthy secondary, for both a 5xx primary and a
+// network-unreachable primary.
+func TestFetchMultiBaseURLFailover(t *testing.T) {
+	requireStack(t)
+	ctx := context.Background()
+
+	var hitsB atomic.Int64
+	upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hitsB.Add(1)
+		w.Write([]byte("served-by-B"))
+	}))
+	defer upB.Close()
+	up500 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer up500.Close()
+
+	p := prov(t, models.PackageGeneric)
+
+	// Primary returns 5xx: every request must still succeed via the secondary.
+	r5xx := seed(t, models.Remote{
+		Name:        "eng-failover-5xx",
+		PackageType: models.PackageGeneric,
+		RepoType:    models.RepoTypeRemote,
+		BaseURL:     up500.URL,
+		Mirrorlist:  []string{upB.URL},
+	})
+	for i := 0; i < 6; i++ {
+		res, err := testEngine.Fetch(ctx, r5xx, fmt.Sprintf("fo5-%d.bin", i), p)
+		if err != nil {
+			t.Fatalf("5xx failover fetch %d: %v", i, err)
+		}
+		if got := readAll(t, res); got != "served-by-B" {
+			t.Fatalf("5xx failover fetch %d body=%q, want served-by-B", i, got)
+		}
+	}
+
+	// Primary is network-unreachable: failover must still reach the secondary.
+	rNet := seed(t, models.Remote{
+		Name:        "eng-failover-net",
+		PackageType: models.PackageGeneric,
+		RepoType:    models.RepoTypeRemote,
+		BaseURL:     "http://127.0.0.1:1",
+		Mirrorlist:  []string{upB.URL},
+	})
+	res, err := testEngine.Fetch(ctx, rNet, "fonet.bin", p)
+	if err != nil {
+		t.Fatalf("network failover fetch: %v", err)
+	}
+	if got := readAll(t, res); got != "served-by-B" {
+		t.Fatalf("network failover body=%q, want served-by-B", got)
+	}
+	if hitsB.Load() == 0 {
+		t.Fatal("secondary upstream never served during failover")
+	}
+}
+
+// TestFetchDefinitiveStatusNoFailover asserts a definitive 404 from the first
+// mirror is returned as-is (not failed over): a missing artifact is not a mirror
+// outage. The remote is fresh so its round-robin cursor starts at index 0.
+func TestFetchDefinitiveStatusNoFailover(t *testing.T) {
+	requireStack(t)
+	ctx := context.Background()
+
+	var hitsB atomic.Int64
+	up404 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.NotFound(w, r)
+	}))
+	defer up404.Close()
+	upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hitsB.Add(1)
+		w.Write([]byte("B"))
+	}))
+	defer upB.Close()
+
+	r := seed(t, models.Remote{
+		Name:        "eng-no-failover-404",
+		PackageType: models.PackageGeneric,
+		RepoType:    models.RepoTypeRemote,
+		BaseURL:     up404.URL,
+		Mirrorlist:  []string{upB.URL},
+	})
+	_, err := testEngine.Fetch(ctx, r, "missing.bin", prov(t, models.PackageGeneric))
+	var pe *ProxyError
+	if err == nil || !asProxyError(err, &pe) || pe.Status != http.StatusNotFound {
+		t.Fatalf("expected 404 ProxyError without failover, got %v", err)
+	}
+	if hitsB.Load() != 0 {
+		t.Fatalf("404 from primary must not fail over, but secondary was hit %d times", hitsB.Load())
+	}
+}
+
+// TestFetchSingleBaseURLUnchanged asserts a single-URL remote behaves exactly as
+// before: one healthy URL succeeds, and one dead URL errors with no failover.
+func TestFetchSingleBaseURLUnchanged(t *testing.T) {
+	requireStack(t)
+	ctx := context.Background()
+
+	upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("solo"))
+	}))
+	defer upB.Close()
+
+	p := prov(t, models.PackageGeneric)
+
+	rOK := seed(t, models.Remote{
+		Name:        "eng-solo",
+		PackageType: models.PackageGeneric,
+		RepoType:    models.RepoTypeRemote,
+		BaseURL:     upB.URL,
+	})
+	res, err := testEngine.Fetch(ctx, rOK, "solo.bin", p)
+	if err != nil {
+		t.Fatalf("single-url fetch: %v", err)
+	}
+	if got := readAll(t, res); got != "solo" {
+		t.Fatalf("single-url body=%q, want solo", got)
+	}
+
+	rDead := seed(t, models.Remote{
+		Name:        "eng-solo-dead",
+		PackageType: models.PackageGeneric,
+		RepoType:    models.RepoTypeRemote,
+		BaseURL:     "http://127.0.0.1:1",
+	})
+	if _, err := testEngine.Fetch(ctx, rDead, "x.bin", p); err == nil {
+		t.Fatal("single dead upstream should error, not succeed")
+	}
+}