From 2ed0327946e10546688d7474bbe981da46ba736a Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 11 Oct 2026 23:29:52 +0200 Subject: [PATCH 1/7] Keep fetching an artifact after the client that started the fetch disconnects. pacman gives up after 10 seconds without data, and since the proxy only starts sending once the whole file is stored, a large package on a slow upstream was cancelled on every try and never got cached. Client requests now mark their context so the shared fetch runs on context.WithoutCancel, still bounded by the artifact client timeout. Mirror jobs don't set the marker, so cancelling a job still aborts its fetches. --- internal/handler/detached_fetch_test.go | 139 ++++++++++++++++++++++++ internal/handler/handler.go | 28 ++++- internal/server/server.go | 5 + 3 files changed, 170 insertions(+), 2 deletions(-) create mode 100644 internal/handler/detached_fetch_test.go diff --git a/internal/handler/detached_fetch_test.go b/internal/handler/detached_fetch_test.go new file mode 100644 index 00000000..2e3196ae --- /dev/null +++ b/internal/handler/detached_fetch_test.go @@ -0,0 +1,139 @@ +package handler + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/registries/fetch" +) + +// gatedFetcher holds every fetch until release is closed, and gives up early +// when the fetch context is cancelled, the way a real HTTP fetch does. +type gatedFetcher struct { + calls atomic.Int64 + content string + entered chan struct{} + release chan struct{} +} + +func newGatedFetcher(content string) *gatedFetcher { + return &gatedFetcher{content: content, entered: make(chan struct{}, 1), release: make(chan struct{})} +} + +func (f *gatedFetcher) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) { + return f.FetchWithHeaders(ctx, url, nil) +} + +func (f *gatedFetcher) FetchWithHeaders(ctx context.Context, _ string, _ http.Header) (*fetch.Artifact, error) { + f.calls.Add(1) + select { + case f.entered <- struct{}{}: + default: + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-f.release: + } + return &fetch.Artifact{ + Body: io.NopCloser(strings.NewReader(f.content)), + ContentType: "application/octet-stream", + }, nil +} + +func (f *gatedFetcher) Head(context.Context, string) (int64, string, error) { + return 0, "", nil +} + +const detachedFetchURL = "https://geo.mirror.pkgbuild.com/extra/os/x86_64/big-1.0-1-x86_64.pkg.tar.zst" + +// startCancelledFetch starts a fetch under ctx, waits until it reached the +// upstream, cancels ctx, lets the upstream answer, and returns the caller's +// error. +func startCancelledFetch(t *testing.T, proxy *Proxy, fetcher *gatedFetcher, ctx context.Context) error { + t.Helper() + ctx, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { + res, err := proxy.GetOrFetchArtifactFromURL(ctx, "arch", "big", "1.0-1", "big-1.0-1-x86_64.pkg.tar.zst", detachedFetchURL) + drain(res) + done <- err + }() + + select { + case <-fetcher.entered: + case <-time.After(5 * time.Second): + t.Fatal("fetch never reached the upstream") + } + cancel() + // Give a non-detached fetch the chance to observe the cancellation before + // the upstream answers. + time.Sleep(50 * time.Millisecond) + close(fetcher.release) + + select { + case err := <-done: + return err + case <-time.After(5 * time.Second): + t.Fatal("fetch did not return") + return nil + } +} + +// TestDetachedFetch_ClientCancelStillCaches covers a client that gives up on a +// slow cache miss. pacman aborts after 10 s without data; if that cancelled +// the upstream fetch, every retry would start from zero and a large package +// would never reach the cache. +func TestDetachedFetch_ClientCancelStillCaches(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + fetcher := newGatedFetcher("package bytes") + proxy.Fetcher = fetcher + + _ = startCancelledFetch(t, proxy, fetcher, DetachUpstreamFetch(context.Background())) + + res, err := proxy.GetOrFetchArtifactFromURL(context.Background(), + "arch", "big", "1.0-1", "big-1.0-1-x86_64.pkg.tar.zst", detachedFetchURL) + if err != nil { + t.Fatalf("follow-up request: %v", err) + } + body, _ := io.ReadAll(res.Reader) + _ = res.Reader.Close() + if !res.Cached { + t.Error("follow-up was not a cache hit: the cancelled client aborted the fetch") + } + if string(body) != "package bytes" { + t.Errorf("body = %q, want %q", body, "package bytes") + } + if got := fetcher.calls.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1", got) + } +} + +// TestDetachedFetch_UnmarkedContextStillCancels keeps mirror jobs cancellable: +// without the marker, cancelling the context aborts the fetch and nothing is +// cached. +func TestDetachedFetch_UnmarkedContextStillCancels(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + fetcher := newGatedFetcher("package bytes") + proxy.Fetcher = fetcher + + err := startCancelledFetch(t, proxy, fetcher, context.Background()) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled", err) + } + + cached, err := proxy.GetCachedArtifact(context.Background(), "arch", "big", "1.0-1", "big-1.0-1-x86_64.pkg.tar.zst") + if err != nil { + t.Fatalf("cache lookup: %v", err) + } + if cached != nil { + _ = cached.Reader.Close() + t.Error("cancelled undetached fetch was still committed to the cache") + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 449fcef7..53445f78 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -612,7 +612,9 @@ type inflightFetch struct { // commit runs on that caller's context, so cancellation behaves as it did // uncoalesced and mirroring still relies on it aborting the fetch. If that // caller goes away, everyone sharing the fetch gets its error and the key is -// released for a later retry. +// released for a later retry. A context marked with DetachUpstreamFetch is the +// exception: there commit keeps running after the caller goes away, so the +// artifact still lands in the cache. // // Every sharing caller still records a cache miss, so the gap between // proxy_cache_misses_total and upstream fetch observations is what coalescing @@ -669,7 +671,11 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch } } - stored, path, err := commit(ctx) + commitCtx := ctx + if upstreamFetchDetached(ctx) { + commitCtx = context.WithoutCancel(ctx) + } + stored, path, err := commit(commitCtx) f.val, f.err = fetchedArtifact{artifact: stored, storagePath: path}, err if err != nil { return nil, err @@ -677,6 +683,24 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch return p.openStoredArtifact(ctx, stored, path) } +type detachUpstreamFetchKey struct{} + +// DetachUpstreamFetch marks ctx so that an artifact fetch started under it +// runs to completion even when ctx is cancelled. The server sets it on client +// requests: a client that gives up on a slow cache miss (pacman aborts after +// 10 s without data) would otherwise cancel the fetch, and every retry would +// start again from zero, so a large artifact never reaches the cache. The +// fetch stays bounded by the artifact HTTP client's timeout. Mirror jobs leave +// it unset so cancelling a job still aborts its fetches. +func DetachUpstreamFetch(ctx context.Context) context.Context { + return context.WithValue(ctx, detachUpstreamFetchKey{}, true) +} + +func upstreamFetchDetached(ctx context.Context) bool { + detached, _ := ctx.Value(detachUpstreamFetchKey{}).(bool) + return detached +} + // fetchedArtifact is what a shared fetch hands its callers: metadata and a // storage path, neither holding reader state. type fetchedArtifact struct { diff --git a/internal/server/server.go b/internal/server/server.go index 380892a7..82c47de5 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -290,6 +290,11 @@ func (s *Server) serve(listener net.Listener) error { next.ServeHTTP(w, r) }) }) + r.Use(func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + next.ServeHTTP(w, r.WithContext(handler.DetachUpstreamFetch(r.Context()))) + }) + }) // Mount protocol handlers s.mountProtocolHandlers(r, proxy) From 111b271273ed8579ee402974c55e1604afe07a9f Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 11 Oct 2026 23:41:16 +0200 Subject: [PATCH 2/7] Arch Linux pacman repositories are now proxied at /arch/{name}/, with geo.mirror.pkgbuild.com as the built-in archlinux repository and more mirrors configurable under upstream.arch. Databases and file lists go through the metadata cache byte for byte, and a database signature is refreshed together with its database so the two always match. Packages and package signatures go to the artifact cache and stay available offline. The path is passed through unchanged, so Arch Linux ARM's $arch/$repo layout works too. README, config example, configuration docs and the dashboard have pacman instructions. --- README.md | 66 ++- config.example.yaml | 9 + docs/configuration.md | 20 + internal/config/config.go | 9 + internal/config/config_test.go | 20 +- internal/handler/arch.go | 218 +++++++++ internal/handler/arch_test.go | 463 +++++++++++++++++++ internal/handler/notfound_ecosystems_test.go | 2 + internal/server/dashboard.go | 15 + internal/server/middleware.go | 2 +- internal/server/middleware_test.go | 2 +- internal/server/server.go | 3 + internal/server/sources.go | 1 + internal/server/sources_test.go | 1 + 14 files changed, 827 insertions(+), 4 deletions(-) create mode 100644 internal/handler/arch.go create mode 100644 internal/handler/arch_test.go diff --git a/README.md b/README.md index 06cd0edb..c3b8571d 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,7 @@ The proxy never uploads artifact bytes to a scanner. Each scanner is notified wi | Debian | Debian/Ubuntu | | ✓ | | RPM | RHEL/Fedora | | ✓ | | Alpine | Alpine Linux | | ✓ | -| Arch | Arch Linux | | ✗ | +| Arch | Arch Linux | | ✓ | | Chef | Chef | | ✗ | | Generic | Any | | ✓ | | Helm | Kubernetes | Yes | ✓ | @@ -665,6 +665,69 @@ http://localhost:8080/apk/private apk appends the architecture and index filename to each repository line itself. +### Arch Linux / pacman + +Put the proxy at the top of `/etc/pacman.d/mirrorlist`. The default repository +name `archlinux` proxies the official GeoIP mirror +(`https://geo.mirror.pkgbuild.com`): + +``` +Server = http://localhost:8080/arch/archlinux/$repo/os/$arch +``` + +Then: + +```bash +pacman -Syu +``` + +Repository databases (`core.db`, `core.files` and their `.tar.*` forms), +database signatures, packages and package signatures are served byte-for-byte +unchanged, so pacman's signature checks keep working. Databases use the +metadata cache (`metadata_ttl`, stale fallback); a database signature is +refreshed together with its database rather than on its own TTL, so the two +always match. Packages and their `.sig` files are stored in the shared artifact +cache and remain available when the upstream is unreachable. + +The proxy only starts sending a package once it has the whole file, and pacman +gives up on a download that sends nothing for 10 seconds. The fetch keeps +running and the next try is served from the cache. On a slow upstream, set +`DisableDownloadTimeout` in `/etc/pacman.conf` to avoid the failed first try. + +`extra.files` is about 50 MB and is read into memory on each `pacman -F` +request. Raise `metadata_max_size` (100 MB by default) for larger custom +repositories. + +A custom repository that rebuilds a package under the same version leaves the +old package and signature in the cache. Bump `pkgrel` instead, or remove the +cached version. + +#### Arch Linux ARM + +Arch Linux ARM has no built-in repository. Add one under `upstream.arch`, using a +regional mirror from that serves HTTPS: + +```yaml +upstream: + arch: + archlinux: "https://geo.mirror.pkgbuild.com" + archlinuxarm: "https://.mirror.archlinuxarm.org" +``` + +Setting `upstream.arch` replaces the built-in default, so keep `archlinux` in the map +if you also proxy x86_64 Arch. + +Arch Linux ARM mirrors use `$arch/$repo` instead of `$repo/os/$arch`. In +`/etc/pacman.d/mirrorlist`: + +``` +Server = http://localhost:8080/arch/archlinuxarm/$arch/$repo +``` + +Avoid `mirror.archlinuxarm.org` as the upstream. It redirects each request to a +changing regional mirror over plain HTTP, so the database and the packages can come +from mirrors with a different sync state. + ### GitHub Releases / mise (aqua backend) Configure named generic upstreams: @@ -993,6 +1056,7 @@ Recently cached: | `GET /v2/*` | OCI/Docker registry protocol | | `GET /v2/homebrew/core/*` | Homebrew core bottle manifests and blobs from GHCR | | `GET /apk/{repository}/*` | Alpine APK repository protocol | +| `GET /arch/{repository}/*` | Arch Linux pacman repository protocol | | `GET /generic/{name}/*` | Generic HTTP download proxy (GitHub release assets, mise/aqua) | | `GET /debian/*` | Debian/APT repository protocol (main archive) | | `GET /debian/{repository}/*` | Debian/APT repository protocol (named archive, e.g. security) | diff --git a/config.example.yaml b/config.example.yaml index d29b4ba7..6704b334 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -252,6 +252,15 @@ upstream: # alpine: "https://dl-cdn.alpinelinux.org/alpine" # private: "https://apk.example.com" + # Named Arch Linux pacman mirrors (used by /arch/{name}/). + # Defaults to {"archlinux": "https://geo.mirror.pkgbuild.com"} when empty; + # configuring any entry replaces that default. Arch Linux ARM needs a + # regional HTTPS mirror of your choice, not the mirror.archlinuxarm.org + # redirector. + # arch: + # archlinux: "https://geo.mirror.pkgbuild.com" + # archlinuxarm: "https://.mirror.archlinuxarm.org" + # Named generic HTTP upstreams (used by /generic/{name}/). The remaining # request path and query are appended to the upstream URL. GitHub release # assets ({owner}/{repo}/releases/download/{tag}/{asset}) are cached diff --git a/docs/configuration.md b/docs/configuration.md index 675d45af..442ec0dc 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -390,6 +390,26 @@ verification keeps working; `.apk` packages use the shared artifact cache. When `upstream.apk` is empty, a single repository named `alpine` pointing at the official mirror is available; configuring any entry replaces that default. +```yaml +upstream: + + # Named Arch Linux pacman mirrors, served at /arch/{name}/. + arch: + archlinux: "https://geo.mirror.pkgbuild.com" +``` + +Arch Linux repositories are read-only. Requests to `/arch/{name}/…` are passed +to the mirror unchanged, so both the Arch Linux layout +(`/arch/archlinux/core/os/x86_64/core.db`) and the Arch Linux ARM layout +(`/arch/archlinuxarm/aarch64/core/core.db`) work. Databases (`{repo}.db`, +`{repo}.files` and their `.tar.*` forms) are cached with the metadata TTL and +served byte-for-byte unchanged; a database's `.sig` is refreshed together with +the database. Packages and package signatures use the shared artifact cache. +When `upstream.arch` is empty, a single repository named `archlinux` pointing at +`https://geo.mirror.pkgbuild.com` is available; configuring any entry replaces +that default. Arch Linux ARM has no default: pick a regional HTTPS mirror, not +the `mirror.archlinuxarm.org` redirector. + ### Debian archives ```yaml diff --git a/internal/config/config.go b/internal/config/config.go index fc314bdf..2e4c850b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -646,6 +646,12 @@ type UpstreamConfig struct { // Default when empty: {"alpine": "https://dl-cdn.alpinelinux.org/alpine"}. APK map[string]string `json:"apk" yaml:"apk"` + // Arch maps repository names to Arch Linux pacman mirror base URLs, + // served at /arch/{name}/. The remaining request path mirrors the + // upstream layout, e.g. /arch/archlinux/core/os/x86_64/core.db. + // Default when empty: {"archlinux": "https://geo.mirror.pkgbuild.com"}. + Arch map[string]string `json:"arch" yaml:"arch"` + // OCI maps names to OCI registry URLs. Requests to a named registry use // the repository prefix upstream/{name}/, for example // oci://proxy.example.com/upstream/ghcr/owner/chart. @@ -717,6 +723,9 @@ func (u *UpstreamConfig) Validate() error { if err := validateNamedUpstreams("upstream.apk", u.APK); err != nil { return err } + if err := validateNamedUpstreams("upstream.arch", u.Arch); err != nil { + return err + } if err := validateNamedUpstreams("upstream.oci", u.OCI); err != nil { return err } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5566afdb..492d4a71 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1364,11 +1364,15 @@ func TestValidateNamedUpstreams(t *testing.T) { wantErr bool }{ { - name: "valid Helm, OCI, APK, and generic upstreams", + name: "valid Helm, OCI, APK, Arch, and generic upstreams", modify: func(cfg *Config) { cfg.Upstream.Helm = map[string]string{"bitnami": "https://charts.bitnami.com/bitnami"} cfg.Upstream.OCI = map[string]string{"ghcr": "https://ghcr.io"} cfg.Upstream.APK = map[string]string{"alpine": "https://dl-cdn.alpinelinux.org/alpine"} + cfg.Upstream.Arch = map[string]string{ + "archlinux": "https://geo.mirror.pkgbuild.com", + "archlinuxarm": "https://de3.mirror.archlinuxarm.org", + } cfg.Upstream.Generic = map[string]string{ "github": "https://github.com", "github-api": "https://api.github.com", @@ -1495,6 +1499,20 @@ func TestValidateNamedUpstreams(t *testing.T) { }, wantErr: true, }, + { + name: "Arch upstream name contains path separator", + modify: func(cfg *Config) { + cfg.Upstream.Arch = map[string]string{"arch/linux": "https://geo.mirror.pkgbuild.com"} + }, + wantErr: true, + }, + { + name: "Arch upstream URL is not absolute", + modify: func(cfg *Config) { + cfg.Upstream.Arch = map[string]string{"archlinux": "geo.mirror.pkgbuild.com"} + }, + wantErr: true, + }, { name: "valid Debian repository", modify: func(cfg *Config) { diff --git a/internal/handler/arch.go b/internal/handler/arch.go new file mode 100644 index 00000000..fd745e62 --- /dev/null +++ b/internal/handler/arch.go @@ -0,0 +1,218 @@ +package handler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "net/http" + "regexp" + "strings" +) + +const ( + archEcosystem = "arch" + defaultArchRepositoryName = "archlinux" + // defaultArchUpstream is the official Arch Linux GeoIP mirror. + defaultArchUpstream = "https://geo.mirror.pkgbuild.com" + archMatchCount = 4 // full match + name + version + arch + archSignatureSuffix = ".sig" + archAccept = "*/*" + archPackageMediaType = "application/octet-stream" +) + +// ArchHandler serves pacman repositories at /arch/{repository}/. The rest of +// the path goes to the mirror unchanged, so the Arch Linux ARM layout works +// too. Databases are served byte for byte through the metadata cache, packages +// and their signatures through the artifact cache. +type ArchHandler struct { + proxy *Proxy + proxyURL string + repositories map[string]string +} + +// NewArchHandler creates an Arch Linux handler. Without repositories it serves +// the official GeoIP mirror as "archlinux". +func NewArchHandler(proxy *Proxy, proxyURL string, repositories map[string]string) *ArchHandler { + h := &ArchHandler{ + proxy: proxy, + proxyURL: strings.TrimSuffix(proxyURL, "/"), + repositories: make(map[string]string, len(repositories)), + } + for name, repositoryURL := range repositories { + h.repositories[name] = strings.TrimSuffix(repositoryURL, "/") + } + if len(h.repositories) == 0 { + h.repositories[defaultArchRepositoryName] = defaultArchUpstream + } + return h +} + +// Routes returns the HTTP handler for pacman requests. +// Mount this at /arch on your router. +func (h *ArchHandler) Routes() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + path := strings.TrimPrefix(r.URL.Path, "/") + + if containsPathTraversal(path) { + http.Error(w, "invalid path", http.StatusBadRequest) + return + } + + repository, rest, ok := strings.Cut(path, "/") + upstreamURL, found := h.repositories[repository] + if !ok || rest == "" || !found { + http.NotFound(w, r) + return + } + + switch { + case isArchDatabase(rest): + h.handleMetadata(w, r, repository, upstreamURL, rest) + case isArchDatabaseSignature(rest): + h.handleDatabaseSignature(w, r, repository, upstreamURL, rest) + case isArchPackage(rest) || isArchPackageSignature(rest): + h.handlePackageDownload(w, r, repository, upstreamURL, rest) + default: + h.proxyFile(w, r, upstreamURL, rest) + } + }) +} + +// archDatabasePattern matches {repo}.db and {repo}.files, bare or as .tar.*. +var archDatabasePattern = regexp.MustCompile(`^[^/]+\.(?:db|files)(?:\.tar(?:\.(?:gz|xz|zst|bz2|lz4|lzo|lrz|lz|Z))?)?$`) + +func archBase(path string) string { + return path[strings.LastIndex(path, "/")+1:] +} + +func isArchDatabase(path string) bool { + return archDatabasePattern.MatchString(archBase(path)) +} + +func isArchDatabaseSignature(path string) bool { + base, ok := strings.CutSuffix(archBase(path), archSignatureSuffix) + return ok && archDatabasePattern.MatchString(base) +} + +// archPackagePattern matches {pkgname}-{pkgver}-{pkgrel}-{arch}.pkg.tar[.ext], +// e.g. lz4-1:1.10.0-2-x86_64.pkg.tar.zst. pkgname may contain '-', and pkgver +// may carry an epoch. +var archPackagePattern = regexp.MustCompile(`^(.+)-((?:\d+:)?[^-/:]+-[^-/:]+)-([^-/:]+)\.pkg\.tar(?:\.(?:zst|xz|gz|bz2|lz4|lzo|lrz|lz|Z))?$`) + +func isArchPackage(path string) bool { + return archPackagePattern.MatchString(archBase(path)) +} + +func isArchPackageSignature(path string) bool { + base, ok := strings.CutSuffix(archBase(path), archSignatureSuffix) + return ok && archPackagePattern.MatchString(base) +} + +// parseArchPackage returns name, pkgver-pkgrel and arch of a package or .sig. +func parseArchPackage(filename string) (name, version, arch string) { + matches := archPackagePattern.FindStringSubmatch(strings.TrimSuffix(filename, archSignatureSuffix)) + if len(matches) != archMatchCount { + return "", "", "" + } + return matches[1], matches[2], matches[3] +} + +// handlePackageDownload caches packages and their signatures. The cache key +// holds repository and full path, since two repositories may ship different +// bytes under one filename. +func (h *ArchHandler) handlePackageDownload(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { + name, version, arch := parseArchPackage(archBase(path)) + if name == "" { + h.proxyFile(w, r, upstreamURL, path) + return + } + + downloadURL := upstreamURL + "/" + path + cacheFilename := repository + "/" + path + + h.proxy.Logger.Info("arch package download", + "repository", repository, "name", name, "version", version, "arch", arch, + "signature", strings.HasSuffix(path, archSignatureSuffix)) + + result, err := h.proxy.GetOrFetchArtifactFromURL( + r.Context(), archEcosystem, name, version, cacheFilename, downloadURL) + if err != nil { + h.proxy.serveArtifactError(w, err, "failed to fetch package") + return + } + + if result.Artifact.MediaType == "" { + result.Artifact.MediaType = archPackageMediaType + } + ServeArtifactRequest(w, r, result) +} + +// handleMetadata serves databases unchanged through the metadata cache. +func (h *ArchHandler) handleMetadata(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { + h.proxy.ProxyCached(w, r, upstreamURL+"/"+path, archEcosystem, + h.metadataCacheKey(repository, upstreamURL, path), archAccept) +} + +// handleDatabaseSignature lets a database's .sig follow the database instead +// of its own TTL, since pacman rejects a mismatched pair. A signature older +// than the cached database is fetched again. +func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { + p := h.proxy + sigKey := h.metadataCacheKey(repository, upstreamURL, path) + if !p.CacheMetadata || p.DB == nil { + h.handleMetadata(w, r, repository, upstreamURL, path) + return + } + + dbKey := h.metadataCacheKey(repository, upstreamURL, strings.TrimSuffix(path, archSignatureSuffix)) + dbEntry, _ := p.DB.GetMetadataCache(archEcosystem, dbKey) + sigEntry, _ := p.DB.GetMetadataCache(archEcosystem, sigKey) + if dbEntry == nil || sigEntry == nil || !dbEntry.FetchedAt.Valid || !sigEntry.FetchedAt.Valid { + h.handleMetadata(w, r, repository, upstreamURL, path) + return + } + + if !sigEntry.FetchedAt.Time.Before(dbEntry.FetchedAt.Time) { + body, contentType, err := p.readCachedMetadata(r.Context(), sigEntry, nil) + if err == nil { + p.writeMetadataCachedResponseWithEncoding(w, r, archEcosystem, sigKey, body, contentType, sigEntry.ContentEncoding.String) + return + } + // The cached bytes are gone; refetch below. + } + + signatureURL := upstreamURL + "/" + path + // Own key: a regular fetch would return the cached copy within its TTL. + key := metadataCoalesceKey(archEcosystem, sigKey, signatureURL, archAccept, "identity", false) + "\x00follow-database" + res, _ := p.coalesceMetadata(r.Context(), key, func(fetchCtx context.Context) metadataResult { + return p.fetchMetadataFromUpstream(fetchCtx, archEcosystem, sigKey, signatureURL, archAccept, "identity", nil, sigEntry) + }) + if res.err != nil { + if errors.Is(res.err, ErrUpstreamNotFound) { + http.Error(w, "not found", http.StatusNotFound) + return + } + p.Logger.Error("metadata fetch failed", "error", res.err) + http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) + return + } + p.writeMetadataCachedResponseWithEncoding(w, r, archEcosystem, sigKey, res.body, res.contentType, res.contentEncoding) +} + +// metadataCacheKey hashes repository, upstream URL and path, as APKHandler does. +func (h *ArchHandler) metadataCacheKey(repository, upstreamURL, path string) string { + identity := repository + "\x00" + upstreamURL + "\x00" + path + digest := sha256.Sum256([]byte(identity)) + return hex.EncodeToString(digest[:]) +} + +// proxyFile passes other files (lastupdate, ISOs) through uncached. +func (h *ArchHandler) proxyFile(w http.ResponseWriter, r *http.Request, upstreamURL, path string) { + h.proxy.ProxyFile(w, r, upstreamURL+"/"+path) +} diff --git a/internal/handler/arch_test.go b/internal/handler/arch_test.go new file mode 100644 index 00000000..778c2dd3 --- /dev/null +++ b/internal/handler/arch_test.go @@ -0,0 +1,463 @@ +package handler + +import ( + "fmt" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/packageurl" + "github.com/git-pkgs/registries/fetch" +) + +func TestParseArchPackage(t *testing.T) { + assertPathParser(t, "parseArchPackage", parseArchPackage, []pathParseCase{ + {"bash-5.2.037-1-x86_64.pkg.tar.zst", "bash", "5.2.037-1", "x86_64"}, + {"python-pip-24.3.1-1-any.pkg.tar.zst", "python-pip", "24.3.1-1", "any"}, + {"lz4-1:1.10.0-2-x86_64.pkg.tar.zst", "lz4", "1:1.10.0-2", "x86_64"}, + {"wpa_supplicant-2:2.12-1-x86_64.pkg.tar.zst", "wpa_supplicant", "2:2.12-1", "x86_64"}, + {"linux-aarch64-6.12.1-1-aarch64.pkg.tar.xz", "linux-aarch64", "6.12.1-1", "aarch64"}, + {"libsigc++-3.6.0-1.1-x86_64.pkg.tar.lz4", "libsigc++", "3.6.0-1.1", "x86_64"}, + {"plain-1.0-1-x86_64.pkg.tar", "plain", "1.0-1", "x86_64"}, + {"bash-5.2.037-1-x86_64.pkg.tar.zst.sig", "bash", "5.2.037-1", "x86_64"}, + {"bash-x86_64.pkg.tar.zst", "", "", ""}, + {"bash-5.2.037-1-x86_64.tar.zst", "", "", ""}, + {"core.db", "", "", ""}, + }) +} + +func TestArchFileClassification(t *testing.T) { + tests := []struct { + path string + database, databaseSig, pkg, pkgSig bool + }{ + {path: "core/os/x86_64/core.db", database: true}, + {path: "core/os/x86_64/core.db.tar.gz", database: true}, + {path: "core/os/x86_64/core.files", database: true}, + {path: "core/os/x86_64/core.files.tar.zst", database: true}, + {path: "aarch64/core/core.db", database: true}, + {path: "core/os/x86_64/core.db.sig", databaseSig: true}, + {path: "custom/os/x86_64/custom.db.tar.xz.sig", databaseSig: true}, + {path: "core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst", pkg: true}, + {path: "core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst.sig", pkgSig: true}, + {path: "core/os/x86_64/core.db.tar.gz.old"}, + {path: "lastupdate"}, + {path: "iso/latest/archlinux-x86_64.iso"}, + } + for _, tt := range tests { + if got := isArchDatabase(tt.path); got != tt.database { + t.Errorf("isArchDatabase(%q) = %v, want %v", tt.path, got, tt.database) + } + if got := isArchDatabaseSignature(tt.path); got != tt.databaseSig { + t.Errorf("isArchDatabaseSignature(%q) = %v, want %v", tt.path, got, tt.databaseSig) + } + if got := isArchPackage(tt.path); got != tt.pkg { + t.Errorf("isArchPackage(%q) = %v, want %v", tt.path, got, tt.pkg) + } + if got := isArchPackageSignature(tt.path); got != tt.pkgSig { + t.Errorf("isArchPackageSignature(%q) = %v, want %v", tt.path, got, tt.pkgSig) + } + } +} + +// TestArchPURL pins the package URL type: git-pkgs/purl maps the "arch" +// ecosystem to pkg:alpm with the "arch" namespace. +func TestArchPURL(t *testing.T) { + pkg, version := packageurl.MakeCacheStrings(archEcosystem, "lz4", "1:1.10.0-2") + if pkg != "pkg:alpm/arch/lz4" { + t.Errorf("package PURL = %q, want %q", pkg, "pkg:alpm/arch/lz4") + } + if version != "pkg:alpm/arch/lz4@1:1.10.0-2" { + t.Errorf("version PURL = %q, want %q", version, "pkg:alpm/arch/lz4@1:1.10.0-2") + } +} + +func TestArchHandler_Routes(t *testing.T) { + h := NewArchHandler(nil, "http://localhost:8080", nil) + assertRoutesBasics(t, h.Routes(), "/archlinux/core/os/x86_64/core.db", "/archlinux/core/../../../etc/passwd") +} + +func TestArchHandler_DefaultsToGeoMirror(t *testing.T) { + h := NewArchHandler(nil, "http://localhost:8080", nil) + if got := h.repositories[defaultArchRepositoryName]; got != defaultArchUpstream { + t.Errorf("default repository = %q, want %q", got, defaultArchUpstream) + } + if len(h.repositories) != 1 { + t.Errorf("repositories = %v, want only the default", h.repositories) + } +} + +func TestArchHandler_UnknownRepositoryReturns404(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + h := NewArchHandler(proxy, "http://localhost:8080", map[string]string{"archlinux": "https://example.test"}) + + for _, target := range []string{"/unknown/core/os/x86_64/core.db", "/archlinux", "/"} { + if w := serveArchRequest(h, http.MethodGet, target, nil); w.Code != http.StatusNotFound { + t.Errorf("%s: status = %d, want 404", target, w.Code) + } + } +} + +// archUpstream is a fake pacman mirror. Files can be swapped and the mirror +// taken offline; requests are counted per path. +type archUpstream struct { + *httptest.Server + mu sync.Mutex + files map[string][]byte + modified map[string]time.Time + hits map[string]int + available atomic.Bool + auth atomic.Value +} + +func newArchUpstream(t *testing.T, files map[string][]byte) *archUpstream { + t.Helper() + u := &archUpstream{files: files, modified: map[string]time.Time{}, hits: map[string]int{}} + u.available.Store(true) + u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !u.available.Load() { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + return + } + u.auth.Store(r.Header.Get("Authorization")) + u.mu.Lock() + data, ok := u.files[r.URL.Path] + modified := u.modified[r.URL.Path] + u.hits[r.URL.Path]++ + u.mu.Unlock() + if !ok { + http.NotFound(w, r) + return + } + if !modified.IsZero() { + w.Header().Set("Last-Modified", modified.UTC().Format(http.TimeFormat)) + } + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write(data) + })) + t.Cleanup(u.Close) + return u +} + +func (u *archUpstream) set(path string, data []byte) { + u.mu.Lock() + defer u.mu.Unlock() + u.files[path] = data +} + +func (u *archUpstream) hitCount(path string) int { + u.mu.Lock() + defer u.mu.Unlock() + return u.hits[path] +} + +func newArchTestProxy(t *testing.T, upstream *archUpstream) *Proxy { + t.Helper() + proxy, _, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.HTTPClient = upstream.Client() + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(upstream.Client()), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + return proxy +} + +func assertArchBody(t *testing.T, w *httptest.ResponseRecorder, label string, want []byte) { + t.Helper() + if w.Code != http.StatusOK { + t.Fatalf("%s: status = %d, want 200: %s", label, w.Code, w.Body.String()) + } + if got := w.Body.String(); got != string(want) { + t.Errorf("%s: body altered:\ngot %q\nwant %q", label, got, want) + } +} + +// TestArchHandler_DatabasesServedUnchanged covers database and file-list +// downloads under the names pacman uses (core.db, core.files) and their +// .tar.* forms: bytes are served unchanged so signatures stay valid, cached +// copies are served within the TTL without contacting the upstream, and the +// cache keeps serving them when the upstream is down. +func TestArchHandler_DatabasesServedUnchanged(t *testing.T) { + files := map[string][]byte{ + "/core/os/x86_64/core.db": []byte("\x1f\x8b\x08core-db-bytes"), + "/core/os/x86_64/core.db.tar.gz": []byte("\x1f\x8b\x08core-db-bytes"), + "/core/os/x86_64/core.db.sig": []byte("core-db-signature"), + "/core/os/x86_64/core.files": []byte("\x1f\x8b\x08core-files-bytes"), + "/core/os/x86_64/core.files.tar.zst": []byte("\x28\xb5\x2f\xfdcore-files-zst"), + } + upstream := newArchUpstream(t, files) + proxy := newArchTestProxy(t, upstream) + proxy.AuthForURL = func(string) (string, string) { return "Authorization", "Bearer arch-token" } + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + // The database goes first, as pacman requests it. + paths := []string{ + "/core/os/x86_64/core.db", "/core/os/x86_64/core.db.sig", "/core/os/x86_64/core.db.tar.gz", + "/core/os/x86_64/core.files", "/core/os/x86_64/core.files.tar.zst", + } + for _, path := range paths { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), path, files[path]) + } + if got, _ := upstream.auth.Load().(string); got != "Bearer arch-token" { + t.Errorf("Authorization = %q, want %q", got, "Bearer arch-token") + } + + for _, path := range paths { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), path+" cached", files[path]) + if got := upstream.hitCount(path); got != 1 { + t.Errorf("%s: upstream hits = %d, want 1 (second read from cache)", path, got) + } + } + + upstream.available.Store(false) + for _, path := range paths { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), path+" offline", files[path]) + } +} + +// TestArchHandler_StaleDatabaseServedWhenUpstreamDown covers the stale +// fallback after the TTL. +func TestArchHandler_StaleDatabaseServedWhenUpstreamDown(t *testing.T) { + db := []byte("core-db-bytes") + upstream := newArchUpstream(t, map[string][]byte{"/core/os/x86_64/core.db": db}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil), "seed", db) + ageMetadata(t, proxy, h, "archlinux", upstream.URL, "core/os/x86_64/core.db", 2*time.Hour) + upstream.available.Store(false) + + w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil) + assertArchBody(t, w, "stale", db) + if got := w.Header().Get("Warning"); got == "" { + t.Error("stale database served without a Warning header") + } +} + +// TestArchHandler_DatabaseIfModifiedSince covers pacman's database refresh, +// which sends If-Modified-Since with the local copy's time. +func TestArchHandler_DatabaseIfModifiedSince(t *testing.T) { + modified := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + upstream := newArchUpstream(t, map[string][]byte{"/core/os/x86_64/core.db": []byte("core-db-bytes")}) + upstream.modified["/core/os/x86_64/core.db"] = modified + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + target := "/archlinux/core/os/x86_64/core.db" + assertArchBody(t, serveArchRequest(h, http.MethodGet, target, nil), "seed", []byte("core-db-bytes")) + + upToDate := http.Header{"If-Modified-Since": {modified.Format(http.TimeFormat)}} + if w := serveArchRequest(h, http.MethodGet, target, upToDate); w.Code != http.StatusNotModified { + t.Errorf("unchanged database: status = %d, want 304", w.Code) + } + older := http.Header{"If-Modified-Since": {modified.Add(-time.Hour).Format(http.TimeFormat)}} + assertArchBody(t, serveArchRequest(h, http.MethodGet, target, older), "older local copy", []byte("core-db-bytes")) +} + +// TestArchHandler_MissingDatabaseSignature covers the official repositories, +// which sign no databases: pacman with DatabaseOptional asks for core.db.sig +// and must get a 404 while the upstream is up. With the upstream down the +// answer is a 502, which pacman also treats as "no signature". +func TestArchHandler_MissingDatabaseSignature(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{"/core/os/x86_64/core.db": []byte("core-db-bytes")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil), "db", []byte("core-db-bytes")) + if w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db.sig", nil); w.Code != http.StatusNotFound { + t.Errorf("missing signature: status = %d, want 404", w.Code) + } + + upstream.available.Store(false) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil), "db offline", []byte("core-db-bytes")) + if w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db.sig", nil); w.Code != http.StatusBadGateway { + t.Errorf("missing signature offline: status = %d, want 502", w.Code) + } +} + +// TestArchHandler_DatabaseSignatureFollowsDatabase covers custom repositories +// with signed databases. The database and its signature are separate cache +// entries; when the database is refreshed, the signature must be refreshed +// with it even inside its own TTL, or pacman would get a new database with +// the old signature. While the database is not refreshed, the cached +// signature keeps matching it and is served without asking the upstream. +func TestArchHandler_DatabaseSignatureFollowsDatabase(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v1", []byte("db v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v1", []byte("sig v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v1 again", []byte("sig v1")) + if got := upstream.hitCount(sigPath); got != 1 { + t.Errorf("signature upstream hits = %d, want 1 while the database is unchanged", got) + } + + // Upstream publishes a new database. The cached database expires before + // the signature does. + upstream.set(dbPath, []byte("db v2")) + upstream.set(sigPath, []byte("sig v2")) + ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:], 2*time.Hour) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v2", []byte("db v2")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig after db refresh", []byte("sig v2")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v2 again", []byte("sig v2")) + if got := upstream.hitCount(sigPath); got != 2 { + t.Errorf("signature upstream hits = %d, want 2", got) + } +} + +// ageMetadata moves a cached metadata entry's fetch time into the past, as if +// it had been fetched age ago. +func ageMetadata(t *testing.T, proxy *Proxy, h *ArchHandler, repository, upstreamURL, path string, age time.Duration) { + t.Helper() + entry, err := proxy.DB.GetMetadataCache(archEcosystem, h.metadataCacheKey(repository, upstreamURL, path)) + if err != nil || entry == nil { + t.Fatalf("metadata entry for %s: %v", path, err) + } + entry.FetchedAt.Time = time.Now().Add(-age) + if err := proxy.DB.UpsertMetadataCache(entry); err != nil { + t.Fatalf("aging metadata entry: %v", err) + } +} + +// TestArchHandler_PackagesAndSignatures covers package and signature +// downloads, cache hits, offline reads and epoch filenames (both as pacman +// sends them and percent-encoded). +func TestArchHandler_PackagesAndSignatures(t *testing.T) { + files := map[string][]byte{ + "/core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst": []byte("bash package"), + "/core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst.sig": []byte("bash signature"), + "/core/os/x86_64/lz4-1:1.10.0-2-x86_64.pkg.tar.zst": []byte("lz4 package"), + "/core/os/x86_64/lz4-1:1.10.0-2-x86_64.pkg.tar.zst.sig": []byte("lz4 signature"), + } + upstream := newArchUpstream(t, files) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + for path, want := range files { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), path, want) + } + + upstream.available.Store(false) + for path, want := range files { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), path+" offline", want) + if got := upstream.hitCount(path); got != 1 { + t.Errorf("%s: upstream hits = %d, want 1", path, got) + } + } + assertArchBody(t, serveArchRequest(h, http.MethodGet, + "/archlinux/core/os/x86_64/lz4-1%3A1.10.0-2-x86_64.pkg.tar.zst", nil), "percent-encoded epoch", files["/core/os/x86_64/lz4-1:1.10.0-2-x86_64.pkg.tar.zst"]) +} + +// TestArchHandler_ArchLinuxARMLayout covers the Arch Linux ARM mirror layout +// ($arch/$repo), served through the same route. +func TestArchHandler_ArchLinuxARMLayout(t *testing.T) { + files := map[string][]byte{ + "/aarch64/core/core.db": []byte("alarm core db"), + "/aarch64/core/bash-5.2.037-1-aarch64.pkg.tar.xz": []byte("alarm bash"), + "/aarch64/core/bash-5.2.037-1-aarch64.pkg.tar.xz.sig": []byte("alarm bash signature"), + } + upstream := newArchUpstream(t, files) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinuxarm": upstream.URL}) + + for path, want := range files { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinuxarm"+path, nil), path, want) + } + upstream.available.Store(false) + for path, want := range files { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinuxarm"+path, nil), path+" offline", want) + } +} + +// TestArchHandler_RepositoriesDoNotShareCache checks that two repositories +// with the same path keep their own bytes, for databases and packages. +func TestArchHandler_RepositoriesDoNotShareCache(t *testing.T) { + const dbPath, pkgPath = "/core/os/x86_64/core.db", "/core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst" + a := newArchUpstream(t, map[string][]byte{dbPath: []byte("db A"), pkgPath: []byte("pkg A")}) + b := newArchUpstream(t, map[string][]byte{dbPath: []byte("db B"), pkgPath: []byte("pkg B")}) + proxy := newArchTestProxy(t, a) + proxy.HTTPClient = http.DefaultClient + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(http.DefaultClient), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"a": a.URL, "b": b.URL}) + + for _, repo := range []string{"a", "b"} { + suffix := map[string]string{"a": "A", "b": "B"}[repo] + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/"+repo+dbPath, nil), repo+" db", []byte("db "+suffix)) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/"+repo+pkgPath, nil), repo+" pkg", []byte("pkg "+suffix)) + } +} + +// TestArchHandler_PackageHead covers HEAD: headers without a body. Range +// requests (pacman resuming a .part download) go through ServeArtifactRequest +// and are covered by its tests on seekable storage. +func TestArchHandler_PackageHead(t *testing.T) { + const path = "/core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst" + pkg := []byte("0123456789 package bytes") + upstream := newArchUpstream(t, map[string][]byte{path: pkg}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), "seed", pkg) + + w := serveArchRequest(h, http.MethodHead, "/archlinux"+path, nil) + if w.Code != http.StatusOK || w.Body.Len() != 0 { + t.Errorf("HEAD: status = %d, body length = %d, want 200 and 0", w.Code, w.Body.Len()) + } + if got := w.Header().Get("Content-Length"); got != fmt.Sprint(len(pkg)) { + t.Errorf("HEAD Content-Length = %q, want %d", got, len(pkg)) + } +} + +// TestArchHandler_PackageDownloadSendsUpstreamAuth covers upstream.auth for +// package downloads; databases are covered in +// TestArchHandler_DatabasesServedUnchanged. +func TestArchHandler_PackageDownloadSendsUpstreamAuth(t *testing.T) { + const path = "/private/os/x86_64/tool-1.0-1-x86_64.pkg.tar.zst" + upstream := newArchUpstream(t, map[string][]byte{path: []byte("private package")}) + proxy := newArchTestProxy(t, upstream) + client := upstream.Client() + client.Transport = &authRoundTripper{base: client.Transport, header: "Authorization", value: "Bearer arch-token"} + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(client), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"private": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/private"+path, nil), path, []byte("private package")) + if got, _ := upstream.auth.Load().(string); got != "Bearer arch-token" { + t.Errorf("Authorization = %q, want %q", got, "Bearer arch-token") + } +} + +// TestArchHandler_OtherFilesProxiedUncached covers files that are neither +// databases nor packages, such as lastupdate. +func TestArchHandler_OtherFilesProxiedUncached(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{"/lastupdate": []byte("1760000000")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + for i := range 2 { + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/lastupdate", nil), fmt.Sprint("lastupdate ", i), []byte("1760000000")) + } + if got := upstream.hitCount("/lastupdate"); got != 2 { + t.Errorf("upstream hits = %d, want 2 (not cached)", got) + } +} + +func serveArchRequest(h *ArchHandler, method, target string, header http.Header) *httptest.ResponseRecorder { + req := httptest.NewRequest(method, target, nil) + for k, v := range header { + req.Header[k] = v + } + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, req) + return w +} diff --git a/internal/handler/notfound_ecosystems_test.go b/internal/handler/notfound_ecosystems_test.go index 9e229fcc..e2141246 100644 --- a/internal/handler/notfound_ecosystems_test.go +++ b/internal/handler/notfound_ecosystems_test.go @@ -21,6 +21,8 @@ func TestArtifactDownloadUpstreamNotFoundReturns404(t *testing.T) { func(p *Proxy) http.Handler { return NewRPMHandler(p, "http://localhost").Routes() }}, {"apk", "/alpine/v3.22/main/x86_64/busybox-1.37.0-r12.apk", func(p *Proxy) http.Handler { return NewAPKHandler(p, "http://localhost", nil).Routes() }}, + {"arch", "/archlinux/core/os/x86_64/bash-5.2.037-1-x86_64.pkg.tar.zst", + func(p *Proxy) http.Handler { return NewArchHandler(p, "http://localhost", nil).Routes() }}, {"nuget", "/v3-flatcontainer/newtonsoft.json/13.0.3/newtonsoft.json.13.0.3.nupkg", func(p *Proxy) http.Handler { return NewNuGetHandler(p, "http://localhost").Routes() }}, {"pypi", "/packages/packages/ab/cd/ef0123456789/requests-2.31.0-py3-none-any.whl", diff --git a/internal/server/dashboard.go b/internal/server/dashboard.go index 05643f66..dbc3a69d 100644 --- a/internal/server/dashboard.go +++ b/internal/server/dashboard.go @@ -134,6 +134,7 @@ func supportedEcosystems() []string { // expected order return []string{ "alpine", + "arch", "cargo", "composer", "conan", @@ -204,6 +205,8 @@ func ecosystemColorClasses(ecosystem string) string { return "bg-amber-100 text-amber-800 dark:bg-amber-900/50 dark:text-amber-300" case "alpine": return "bg-lime-100 text-lime-800 dark:bg-lime-900/50 dark:text-lime-300" + case "arch": + return "bg-blue-100 text-blue-800 dark:bg-blue-900/50 dark:text-blue-300" default: return "bg-gray-100 text-gray-700 dark:bg-gray-800 dark:text-gray-300" } @@ -521,5 +524,17 @@ sudo dnf update`), # Then run: apk update`), }, + { + ID: "arch", + Name: "Arch Linux pacman", + Language: "Arch Linux", + Endpoint: "/arch/", + Instructions: template.HTML(`

Configure pacman to use the proxy:

+
# In /etc/pacman.d/mirrorlist, above any other Server line
+Server = ` + baseURL + `/arch/archlinux/$repo/os/$arch
+
+# Then run:
+pacman -Syu
`), + }, } } diff --git a/internal/server/middleware.go b/internal/server/middleware.go index a5433ce5..eabe9f49 100644 --- a/internal/server/middleware.go +++ b/internal/server/middleware.go @@ -106,7 +106,7 @@ func requestEcosystem(path string) string { segment, _, _ := strings.Cut(strings.TrimPrefix(path, "/"), "/") switch segment { case "npm", "cargo", "hex", "pub", "pypi", "maven", "gradle", "nuget", - "conan", "conda", "cran", "julia", "debian", "rpm", + "conan", "conda", "cran", "julia", "debian", "rpm", "arch", "helm", "homebrew", "generic", "swift": return segment case "apk": diff --git a/internal/server/middleware_test.go b/internal/server/middleware_test.go index 595d7ca3..875b152e 100644 --- a/internal/server/middleware_test.go +++ b/internal/server/middleware_test.go @@ -313,7 +313,7 @@ func TestRequestEcosystemCoversEveryMountedRoute(t *testing.T) { "/gradle/x": "gradle", "/nuget/x": "nuget", "/composer/x": "packagist", "/conan/x": "conan", "/conda/x": "conda", "/cran/x": "cran", "/julia/x": "julia", "/swift/x": "swift", "/v2/x": "oci", - "/apk/x": "alpine", "/debian/x": "debian", "/rpm/x": "rpm", + "/apk/x": "alpine", "/arch/x": "arch", "/debian/x": "debian", "/rpm/x": "rpm", "/helm/x": "helm", "/homebrew/x": "homebrew", "/generic/x": "generic", } for path, want := range mounts { diff --git a/internal/server/server.go b/internal/server/server.go index 82c47de5..e5df6246 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -19,6 +19,7 @@ // - /swift/* - Swift Package Registry protocol // - /v2/* - OCI/Docker container registry protocol // - /apk/* - Alpine APK repository protocol +// - /arch/* - Arch Linux pacman repository protocol // - /debian/* - Debian/APT repository protocol // - /rpm/* - RPM/Yum repository protocol // @@ -449,6 +450,7 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { helmHandler := handler.NewHelmHandlerWithOCIRegistries( proxy, s.cfg.BaseURL, s.cfg.Upstream.Helm, s.cfg.Upstream.OCIDefault, s.cfg.Upstream.OCI) apkHandler := handler.NewAPKHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.APK) + archHandler := handler.NewArchHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.Arch) debianHandler := handler.NewDebianHandler( proxy, s.cfg.BaseURL, @@ -478,6 +480,7 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { r.Mount("/v2", http.StripPrefix("/v2", containerHandler.Routes())) r.Mount("/helm", http.StripPrefix("/helm", helmHandler.Routes())) r.Mount("/apk", http.StripPrefix("/apk", apkHandler.Routes())) + r.Mount("/arch", http.StripPrefix("/arch", archHandler.Routes())) r.Mount("/debian", http.StripPrefix("/debian", debianHandler.Routes())) r.Mount("/rpm", http.StripPrefix("/rpm", rpmHandler.Routes())) r.Mount("/generic", http.StripPrefix("/generic", genericHandler.Routes())) diff --git a/internal/server/sources.go b/internal/server/sources.go index 923cb9dc..2a0e47ce 100644 --- a/internal/server/sources.go +++ b/internal/server/sources.go @@ -64,6 +64,7 @@ var knownClients = map[string]string{ "dnf": "dnf", "urlgrabber": "dnf", "apk": "apk", + "pacman": "pacman", "conda": "conda", "mamba": "conda", "conan": "conan", diff --git a/internal/server/sources_test.go b/internal/server/sources_test.go index fa58dd73..560ca93d 100644 --- a/internal/server/sources_test.go +++ b/internal/server/sources_test.go @@ -21,6 +21,7 @@ func TestClientName(t *testing.T) { {"docker/24.0.7 go/go1.20.10 kernel/6.5.0 os/linux", "docker"}, {"Debian APT-HTTP/1.3 (2.6.1)", "apt"}, {"bundler/2.4.22 rubygems/3.4.22", "bundler"}, + {"pacman/7.0.0 (Linux x86_64) libalpm/15.0.0", "pacman"}, // Unknown and hostile input collapses, so a User-Agent cannot mint a // new Prometheus time series. {"", unknownClient}, From 094a962697f80ba757070d8e7638ab1aa6495de6 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 11 Oct 2026 23:51:43 +0200 Subject: [PATCH 3/7] At most 64 fetches keep running after their client hangs up; beyond that a fetch is tied to its client again, so nobody can start an unbounded number of large downloads and leave. A server test now checks that client requests actually get the detach marker. --- internal/handler/detached_fetch_test.go | 26 +++++ internal/handler/handler.go | 40 +++++++- internal/server/detached_fetch_test.go | 122 ++++++++++++++++++++++++ 3 files changed, 185 insertions(+), 3 deletions(-) create mode 100644 internal/server/detached_fetch_test.go diff --git a/internal/handler/detached_fetch_test.go b/internal/handler/detached_fetch_test.go index 2e3196ae..554761d9 100644 --- a/internal/handler/detached_fetch_test.go +++ b/internal/handler/detached_fetch_test.go @@ -137,3 +137,29 @@ func TestDetachedFetch_UnmarkedContextStillCancels(t *testing.T) { t.Error("cancelled undetached fetch was still committed to the cache") } } + +// TestDetachedFetch_CapFallsBackToClientContext checks the cap on detached +// fetches: with every slot taken, a cancelled client cancels its fetch again, +// as it did before fetches were detached. +func TestDetachedFetch_CapFallsBackToClientContext(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + fetcher := newGatedFetcher("package bytes") + proxy.Fetcher = fetcher + + for range maxDetachedFetches { + if !proxy.acquireDetachSlot() { + t.Fatal("could not take a free detach slot") + } + } + err := startCancelledFetch(t, proxy, fetcher, DetachUpstreamFetch(context.Background())) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled once the detach slots are used up", err) + } + + for range maxDetachedFetches { + proxy.releaseDetachSlot() + } + if !proxy.acquireDetachSlot() { + t.Error("slots not available again after release") + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 53445f78..25526928 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -198,6 +198,11 @@ type Proxy struct { fetchMu sync.Mutex inFlight map[string]*inflightFetch + // detachSlots caps the artifact fetches that keep running after their + // client went away (see DetachUpstreamFetch). Guarded by fetchMu while it + // is created. + detachSlots chan struct{} + // inFlightMeta does the same for metadata misses. Keyed by // metadataCoalesceKey. metaMu sync.Mutex @@ -672,7 +677,8 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch } commitCtx := ctx - if upstreamFetchDetached(ctx) { + if upstreamFetchDetached(ctx) && p.acquireDetachSlot() { + defer p.releaseDetachSlot() commitCtx = context.WithoutCancel(ctx) } stored, path, err := commit(commitCtx) @@ -690,8 +696,9 @@ type detachUpstreamFetchKey struct{} // requests: a client that gives up on a slow cache miss (pacman aborts after // 10 s without data) would otherwise cancel the fetch, and every retry would // start again from zero, so a large artifact never reaches the cache. The -// fetch stays bounded by the artifact HTTP client's timeout. Mirror jobs leave -// it unset so cancelling a job still aborts its fetches. +// fetch stays bounded by the artifact HTTP client's timeout, and at most +// maxDetachedFetches run detached at once. Mirror jobs leave it unset so +// cancelling a job still aborts its fetches. func DetachUpstreamFetch(ctx context.Context) context.Context { return context.WithValue(ctx, detachUpstreamFetchKey{}, true) } @@ -701,6 +708,33 @@ func upstreamFetchDetached(ctx context.Context) bool { return detached } +// maxDetachedFetches caps the artifact fetches that may outlive their client. +// Without a cap a client could start any number of large uncached downloads +// and hang up straight away, leaving each to run to the artifact client's +// timeout. Fetches beyond the cap stay tied to their client, as before +// detaching existed. +const maxDetachedFetches = 64 + +func (p *Proxy) acquireDetachSlot() bool { + p.fetchMu.Lock() + if p.detachSlots == nil { + p.detachSlots = make(chan struct{}, maxDetachedFetches) + } + slots := p.detachSlots + p.fetchMu.Unlock() + + select { + case slots <- struct{}{}: + return true + default: + return false + } +} + +func (p *Proxy) releaseDetachSlot() { + <-p.detachSlots +} + // fetchedArtifact is what a shared fetch hands its callers: metadata and a // storage path, neither holding reader state. type fetchedArtifact struct { diff --git a/internal/server/detached_fetch_test.go b/internal/server/detached_fetch_test.go new file mode 100644 index 00000000..557de510 --- /dev/null +++ b/internal/server/detached_fetch_test.go @@ -0,0 +1,122 @@ +package server + +import ( + "context" + "errors" + "io" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/config" +) + +// TestServerKeepsFetchAfterClientHangsUp checks that the server marks client +// requests with handler.DetachUpstreamFetch: a pacman client that gives up on +// a slow cache miss must not cancel the upstream fetch, so the next request +// is served from the cache without fetching again. +func TestServerKeepsFetchAfterClientHangsUp(t *testing.T) { + if os.Getenv("PROXY_TEST_DETACHED_FETCH") == "1" { + testServerKeepsFetchAfterClientHangsUp(t) + return + } + + cmd := exec.Command(os.Args[0], "-test.run=^TestServerKeepsFetchAfterClientHangsUp$") + cmd.Env = append(os.Environ(), "PROXY_TEST_DETACHED_FETCH=1") + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("detached fetch test failed: %v\n%s", err, output) + } +} + +func testServerKeepsFetchAfterClientHangsUp(t *testing.T) { + const pkgPath = "/core/os/x86_64/big-1.0-1-x86_64.pkg.tar.zst" + release := make(chan struct{}) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) + var hits atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != pkgPath { + http.NotFound(w, r) + return + } + hits.Add(1) + <-release + _, _ = io.WriteString(w, "big package") + })) + defer upstream.Close() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("reserving proxy address: %v", err) + } + t.Cleanup(func() { _ = listener.Close() }) + + tempDir := t.TempDir() + cfg := config.Default() + cfg.Listen = listener.Addr().String() + cfg.BaseURL = "http://" + cfg.Listen + cfg.Database.Path = filepath.Join(tempDir, "proxy.db") + cfg.Storage.URL = "file://" + filepath.Join(tempDir, "artifacts") + cfg.Upstream.Arch = map[string]string{"archlinux": upstream.URL} + cfg.Upstream.AllowLoopback = true + if err := cfg.Validate(); err != nil { + t.Fatalf("validating config: %v", err) + } + + proxyServer, err := New(cfg, slog.New(slog.NewTextHandler(io.Discard, nil)), BuildInfo{Version: "test", Commit: "test"}) + if err != nil { + t.Fatalf("creating server: %v", err) + } + startErr := make(chan error, 1) + go func() { startErr <- proxyServer.Start(listener) }() + defer func() { + releaseOnce.Do(func() { close(release) }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := proxyServer.Shutdown(ctx); err != nil { + t.Errorf("shutting down server: %v", err) + } + if err := <-startErr; !errors.Is(err, http.ErrServerClosed) { + t.Errorf("Start() error = %v, want %v", err, http.ErrServerClosed) + } + }() + + target := cfg.BaseURL + "/arch/archlinux" + pkgPath + + // The client gives up while the upstream is still sending nothing. + impatient := &http.Client{Timeout: 300 * time.Millisecond} + deadline := time.Now().Add(5 * time.Second) + for hits.Load() == 0 { + if _, err := impatient.Get(target); err == nil { + t.Fatal("request succeeded while the upstream was held") + } + if time.Now().After(deadline) { + t.Fatal("upstream fetch never started") + } + } + // Let the server notice the hang-up before the upstream answers. + time.Sleep(100 * time.Millisecond) + releaseOnce.Do(func() { close(release) }) + + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Get(target) + if err != nil { + t.Fatalf("follow-up request: %v", err) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK || string(body) != "big package" { + t.Fatalf("follow-up: status %d, body %q", resp.StatusCode, body) + } + if got := hits.Load(); got != 1 { + t.Errorf("upstream fetches = %d, want 1: the client hanging up cancelled the first fetch", got) + } +} From 5ce445f5d3f64ba2bf49a785c745abd223bc2e91 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 11 Oct 2026 23:51:43 +0200 Subject: [PATCH 4/7] A database signature that belongs to an older database is no longer served. When the database was refreshed and the signature refetch fails, the proxy answers 404 or 502 instead of falling back to the cached signature, which pacman would reject. Paths with a decoded ? or # are refused, since they would fetch one file and cache it under another name. The README now says that pacman -Sy only works offline with cache_metadata enabled. --- README.md | 6 ++-- internal/handler/arch.go | 9 +++-- internal/handler/arch_test.go | 66 +++++++++++++++++++++++++++++++++++ 3 files changed, 76 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index c3b8571d..ad478059 100644 --- a/README.md +++ b/README.md @@ -685,8 +685,10 @@ Repository databases (`core.db`, `core.files` and their `.tar.*` forms), database signatures, packages and package signatures are served byte-for-byte unchanged, so pacman's signature checks keep working. Databases use the metadata cache (`metadata_ttl`, stale fallback); a database signature is -refreshed together with its database rather than on its own TTL, so the two -always match. Packages and their `.sig` files are stored in the shared artifact +refreshed together with its database rather than on its own TTL, and a +signature that belongs to an older database is never served in its place. Enable `cache_metadata` (or set `PROXY_CACHE_METADATA=true`) so +`pacman -Sy` keeps working while the upstream is down; without it databases are +fetched fresh on every request. Packages and their `.sig` files are stored in the shared artifact cache and remain available when the upstream is unreachable. The proxy only starts sending a package once it has the whole file, and pacman diff --git a/internal/handler/arch.go b/internal/handler/arch.go index fd745e62..a38fbc1f 100644 --- a/internal/handler/arch.go +++ b/internal/handler/arch.go @@ -59,7 +59,9 @@ func (h *ArchHandler) Routes() http.Handler { path := strings.TrimPrefix(r.URL.Path, "/") - if containsPathTraversal(path) { + // A decoded '?' or '#' would cut the upstream path short and cache one + // file under another file's name. + if containsPathTraversal(path) || strings.ContainsAny(path, "?#") { http.Error(w, "invalid path", http.StatusBadRequest) return } @@ -161,7 +163,7 @@ func (h *ArchHandler) handleMetadata(w http.ResponseWriter, r *http.Request, rep // handleDatabaseSignature lets a database's .sig follow the database instead // of its own TTL, since pacman rejects a mismatched pair. A signature older -// than the cached database is fetched again. +// than the cached database is fetched again and never served as a fallback. func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { p := h.proxy sigKey := h.metadataCacheKey(repository, upstreamURL, path) @@ -191,7 +193,8 @@ func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Req // Own key: a regular fetch would return the cached copy within its TTL. key := metadataCoalesceKey(archEcosystem, sigKey, signatureURL, archAccept, "identity", false) + "\x00follow-database" res, _ := p.coalesceMetadata(r.Context(), key, func(fetchCtx context.Context) metadataResult { - return p.fetchMetadataFromUpstream(fetchCtx, archEcosystem, sigKey, signatureURL, archAccept, "identity", nil, sigEntry) + // No entry: a 304 or stale fallback would return the old signature. + return p.fetchMetadataFromUpstream(fetchCtx, archEcosystem, sigKey, signatureURL, archAccept, "identity", nil, nil) }) if res.err != nil { if errors.Is(res.err, ErrUpstreamNotFound) { diff --git a/internal/handler/arch_test.go b/internal/handler/arch_test.go index 778c2dd3..587ab340 100644 --- a/internal/handler/arch_test.go +++ b/internal/handler/arch_test.go @@ -312,6 +312,72 @@ func TestArchHandler_DatabaseSignatureFollowsDatabase(t *testing.T) { } } +// TestArchHandler_OutdatedSignatureNeverServed covers a signature refetch that +// fails after the database was refreshed. The cached signature belongs to +// the old database, and pacman rejects a present but invalid signature even +// with DatabaseOptional, so the proxy must answer 404 or 502 rather than fall +// back to it. +func TestArchHandler_OutdatedSignatureNeverServed(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + + for _, tt := range []struct { + name string + breakSig func(u *archUpstream) + want int + }{ + {"signature removed upstream", func(u *archUpstream) { + u.mu.Lock() + delete(u.files, sigPath) + u.mu.Unlock() + }, http.StatusNotFound}, + {"upstream down after database refresh", func(u *archUpstream) { u.available.Store(false) }, http.StatusBadGateway}, + } { + t.Run(tt.name, func(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v1", []byte("db v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v1", []byte("sig v1")) + + upstream.set(dbPath, []byte("db v2")) + ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:], 2*time.Hour) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v2", []byte("db v2")) + + tt.breakSig(upstream) + for i := range 2 { + w := serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil) + if w.Code != tt.want { + t.Errorf("request %d: status = %d, body = %q, want %d", i, w.Code, w.Body.String(), tt.want) + } + } + }) + } +} + +// TestArchHandler_RejectsQueryAndFragmentInPath covers a decoded '?' or '#' +// in the path, which would otherwise fetch one file and cache it under the +// name that follows. +func TestArchHandler_RejectsQueryAndFragmentInPath(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + for _, target := range []string{ + "/archlinux/core/os/x86_64/bash-5.2-1-x86_64.pkg.tar.zst%3F/foo-9-9-any.pkg.tar.zst", + "/archlinux/core/os/x86_64/bash-5.2-1-x86_64.pkg.tar.zst%23/foo-9-9-any.pkg.tar.zst", + } { + if w := serveArchRequest(h, http.MethodGet, target, nil); w.Code != http.StatusBadRequest { + t.Errorf("%s: status = %d, want 400", target, w.Code) + } + } + upstream.mu.Lock() + defer upstream.mu.Unlock() + if len(upstream.hits) != 0 { + t.Errorf("upstream was asked for %v", upstream.hits) + } +} + // ageMetadata moves a cached metadata entry's fetch time into the past, as if // it had been fetched age ago. func ageMetadata(t *testing.T, proxy *Proxy, h *ArchHandler, repository, upstreamURL, path string, age time.Duration) { From 086abebe6eff924276cec5e5ab55fb2250bb1a72 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 10:59:22 +0200 Subject: [PATCH 5/7] A database refresh now fetches the signature as well and stores it for that exact database content, keyed by its digest. A signature requested later, after the upstream has moved on, is still the one for the database the client got, and a refresh that returns the same bytes keeps the cached signature, so it is served offline too. The database is fetched once more after its signature to catch a mirror that swapped both in between. A missing signature is remembered per database, so official repos answer 404 without going upstream. --- README.md | 6 +- internal/handler/arch.go | 206 +++++++++++++++++++++++++++++----- internal/handler/arch_test.go | 134 +++++++++++++++++----- 3 files changed, 289 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index ad478059..2f839cca 100644 --- a/README.md +++ b/README.md @@ -684,9 +684,9 @@ pacman -Syu Repository databases (`core.db`, `core.files` and their `.tar.*` forms), database signatures, packages and package signatures are served byte-for-byte unchanged, so pacman's signature checks keep working. Databases use the -metadata cache (`metadata_ttl`, stale fallback); a database signature is -refreshed together with its database rather than on its own TTL, and a -signature that belongs to an older database is never served in its place. Enable `cache_metadata` (or set `PROXY_CACHE_METADATA=true`) so +metadata cache (`metadata_ttl`, stale fallback). Each database refresh also +fetches its signature and stores it for exactly that database, so a signature +is only ever served with the database it signs. Enable `cache_metadata` (or set `PROXY_CACHE_METADATA=true`) so `pacman -Sy` keeps working while the upstream is down; without it databases are fetched fresh on every request. Packages and their `.sig` files are stored in the shared artifact cache and remain available when the upstream is unreachable. diff --git a/internal/handler/arch.go b/internal/handler/arch.go index a38fbc1f..f3505a42 100644 --- a/internal/handler/arch.go +++ b/internal/handler/arch.go @@ -8,6 +8,9 @@ import ( "net/http" "regexp" "strings" + + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/metrics" ) const ( @@ -75,7 +78,7 @@ func (h *ArchHandler) Routes() http.Handler { switch { case isArchDatabase(rest): - h.handleMetadata(w, r, repository, upstreamURL, rest) + h.handleDatabase(w, r, repository, upstreamURL, rest) case isArchDatabaseSignature(rest): h.handleDatabaseSignature(w, r, repository, upstreamURL, rest) case isArchPackage(rest) || isArchPackageSignature(rest): @@ -161,51 +164,204 @@ func (h *ArchHandler) handleMetadata(w http.ResponseWriter, r *http.Request, rep h.metadataCacheKey(repository, upstreamURL, path), archAccept) } -// handleDatabaseSignature lets a database's .sig follow the database instead -// of its own TTL, since pacman rejects a mismatched pair. A signature older -// than the cached database is fetched again and never served as a fallback. -func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { +// archDatabase names a database and its signature in the cache. +type archDatabase struct { + dbKey, dbURL string + sigKey, sigURL string +} + +func (h *ArchHandler) database(repository, upstreamURL, path string) archDatabase { + return archDatabase{ + dbKey: h.metadataCacheKey(repository, upstreamURL, path), + dbURL: upstreamURL + "/" + path, + sigKey: h.metadataCacheKey(repository, upstreamURL, path+archSignatureSuffix), + sigURL: upstreamURL + "/" + path + archSignatureSuffix, + } +} + +func (d archDatabase) coalesceKey(kind string) string { + return metadataCoalesceKey(archEcosystem, d.dbKey, d.dbURL, archAccept, "identity", false) + "\x00" + kind +} + +// handleDatabase serves a database. Each refresh fetches the signature too, so +// a signature requested later still matches the database the client got. +func (h *ArchHandler) handleDatabase(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { p := h.proxy - sigKey := h.metadataCacheKey(repository, upstreamURL, path) if !p.CacheMetadata || p.DB == nil { h.handleMetadata(w, r, repository, upstreamURL, path) return } + d := h.database(repository, upstreamURL, path) + if _, hit := p.cachedMetadataState(r.Context(), archEcosystem, d.dbKey, nil); hit != nil { + metrics.RecordCacheHit(archEcosystem) + h.writeMetadata(w, r, d.dbKey, *hit) + return + } + p.recordMetadataCacheMiss(archEcosystem) + res, _ := p.coalesceMetadata(r.Context(), d.coalesceKey("database"), func(ctx context.Context) metadataResult { + entry, hit := p.cachedMetadataState(ctx, archEcosystem, d.dbKey, nil) + if hit != nil { + return *hit + } + res, _ := h.refresh(ctx, d, entry) + return res + }) + h.writeMetadata(w, r, d.dbKey, res) +} - dbKey := h.metadataCacheKey(repository, upstreamURL, strings.TrimSuffix(path, archSignatureSuffix)) - dbEntry, _ := p.DB.GetMetadataCache(archEcosystem, dbKey) - sigEntry, _ := p.DB.GetMetadataCache(archEcosystem, sigKey) - if dbEntry == nil || sigEntry == nil || !dbEntry.FetchedAt.Valid || !sigEntry.FetchedAt.Valid { +// handleDatabaseSignature serves the signature stored for the cached +// database's content, never one fetched for another version of it. +func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { + p := h.proxy + if !p.CacheMetadata || p.DB == nil { h.handleMetadata(w, r, repository, upstreamURL, path) return } + d := h.database(repository, upstreamURL, strings.TrimSuffix(path, archSignatureSuffix)) + if res, ok := h.cachedSignature(r.Context(), d); ok { + metrics.RecordCacheHit(archEcosystem) + h.writeMetadata(w, r, d.sigKey, res) + return + } + p.recordMetadataCacheMiss(archEcosystem) + // Nothing stored for this database yet, e.g. the signature fetch failed + // during the last refresh: refresh both. + res, _ := p.coalesceMetadata(r.Context(), d.coalesceKey("signature"), func(ctx context.Context) metadataResult { + if res, ok := h.cachedSignature(ctx, d); ok { + return res + } + entry, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) + if _, err := h.refresh(ctx, d, entry); err != nil { + return metadataResult{err: err} + } + if res, ok := h.cachedSignature(ctx, d); ok { + return res + } + return metadataResult{err: errArchSignatureUnpaired} + }) + h.writeMetadata(w, r, d.sigKey, res) +} - if !sigEntry.FetchedAt.Time.Before(dbEntry.FetchedAt.Time) { - body, contentType, err := p.readCachedMetadata(r.Context(), sigEntry, nil) - if err == nil { - p.writeMetadataCachedResponseWithEncoding(w, r, archEcosystem, sigKey, body, contentType, sigEntry.ContentEncoding.String) - return +var errArchSignatureUnpaired = errors.New("database signature could not be paired with the database") + +// archPairAttempts bounds refetches while the mirror replaces the database. +const archPairAttempts = 3 + +// refresh fetches the database, then its signature, and stores the signature +// for the database's content. It returns the database and why no signature +// could be stored, if so. +func (h *ArchHandler) refresh(ctx context.Context, d archDatabase, entry *database.MetadataCacheEntry) (metadataResult, error) { + p := h.proxy + res := p.fetchMetadataFromUpstream(ctx, archEcosystem, d.dbKey, d.dbURL, archAccept, "identity", nil, entry) + if res.err != nil { + return res, res.err + } + for range archPairAttempts { + current, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) + if current == nil || !current.ContentDigest.Valid { + return res, errArchSignatureUnpaired } - // The cached bytes are gone; refetch below. + dbDigest := current.ContentDigest.String + if _, ok := h.storedSignature(ctx, d, dbDigest); ok { + return res, nil + } + + sig, err := p.fetchUpstreamMetadata(ctx, d.sigURL, nil, archAccept, "identity") + if errors.Is(err, ErrUpstreamNotFound) { + h.storeSignature(ctx, d, dbDigest, nil) + return res, err + } + if err != nil { + return res, err + } + + // The mirror may have replaced the database in the meantime. + again, err := p.fetchUpstreamMetadata(ctx, d.dbURL, current, archAccept, "identity") + if err != nil { + return res, err + } + if archDigest(again.body) == dbDigest { + h.storeSignature(ctx, d, dbDigest, sig) + return res, nil + } + p.cacheMetadataBlob(ctx, archEcosystem, d.dbKey, metadataStoragePath(archEcosystem, d.dbKey), again) + res = metadataResult{body: again.body, contentType: again.contentType, contentEncoding: again.contentEncoding} } + return res, errArchSignatureUnpaired +} - signatureURL := upstreamURL + "/" + path - // Own key: a regular fetch would return the cached copy within its TTL. - key := metadataCoalesceKey(archEcosystem, sigKey, signatureURL, archAccept, "identity", false) + "\x00follow-database" - res, _ := p.coalesceMetadata(r.Context(), key, func(fetchCtx context.Context) metadataResult { - // No entry: a 304 or stale fallback would return the old signature. - return p.fetchMetadataFromUpstream(fetchCtx, archEcosystem, sigKey, signatureURL, archAccept, "identity", nil, nil) - }) +// cachedSignature returns the signature stored for the cached database. +func (h *ArchHandler) cachedSignature(ctx context.Context, d archDatabase) (metadataResult, bool) { + current, _ := h.proxy.DB.GetMetadataCache(archEcosystem, d.dbKey) + if current == nil || !current.ContentDigest.Valid { + return metadataResult{}, false + } + return h.storedSignature(ctx, d, current.ContentDigest.String) +} + +// storedSignature returns the signature stored for dbDigest. The storage +// path names the digest, so a refresh with unchanged bytes keeps it. +func (h *ArchHandler) storedSignature(ctx context.Context, d archDatabase, dbDigest string) (metadataResult, bool) { + entry, _ := h.proxy.DB.GetMetadataCache(archEcosystem, d.sigKey) + if entry == nil { + return metadataResult{}, false + } + switch entry.StoragePath { + case archMissingSignaturePath(d, dbDigest): + return metadataResult{err: ErrUpstreamNotFound}, true + case archSignaturePath(d, dbDigest): + body, contentType, err := h.proxy.readCachedMetadata(ctx, entry, nil) + if err != nil { + return metadataResult{}, false + } + return metadataResult{body: body, contentType: contentType, contentEncoding: entry.ContentEncoding.String}, true + } + return metadataResult{}, false +} + +// storeSignature stores sig for dbDigest, or that it is missing when sig is +// nil, and drops the one stored for an earlier database. +func (h *ArchHandler) storeSignature(ctx context.Context, d archDatabase, dbDigest string, sig *upstreamMetadata) { + p := h.proxy + path := archSignaturePath(d, dbDigest) + if sig == nil { + path = archMissingSignaturePath(d, dbDigest) + sig = &upstreamMetadata{contentType: archPackageMediaType} + } + previous, _ := p.DB.GetMetadataCache(archEcosystem, d.sigKey) + p.cacheMetadataBlob(ctx, archEcosystem, d.sigKey, path, sig) + if previous == nil || previous.StoragePath == path { + return + } + if current, _ := p.DB.GetMetadataCache(archEcosystem, d.sigKey); current != nil && current.StoragePath == path { + _ = p.Storage.Delete(ctx, previous.StoragePath) + } +} + +func archSignaturePath(d archDatabase, dbDigest string) string { + return "_metadata/" + archEcosystem + "/" + d.sigKey + "/" + strings.TrimPrefix(dbDigest, "sha256:") +} + +func archMissingSignaturePath(d archDatabase, dbDigest string) string { + return archSignaturePath(d, dbDigest) + "-missing" +} + +func archDigest(body []byte) string { + digest := sha256.Sum256(body) + return "sha256:" + hex.EncodeToString(digest[:]) +} + +func (h *ArchHandler) writeMetadata(w http.ResponseWriter, r *http.Request, cacheKey string, res metadataResult) { if res.err != nil { if errors.Is(res.err, ErrUpstreamNotFound) { http.Error(w, "not found", http.StatusNotFound) return } - p.Logger.Error("metadata fetch failed", "error", res.err) + h.proxy.Logger.Error("metadata fetch failed", "error", res.err) http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return } - p.writeMetadataCachedResponseWithEncoding(w, r, archEcosystem, sigKey, res.body, res.contentType, res.contentEncoding) + h.proxy.writeMetadataCachedResponseWithEncoding(w, r, archEcosystem, cacheKey, res.body, res.contentType, res.contentEncoding) } // metadataCacheKey hashes repository, upstream URL and path, as APKHandler does. diff --git a/internal/handler/arch_test.go b/internal/handler/arch_test.go index 587ab340..1b71b3f7 100644 --- a/internal/handler/arch_test.go +++ b/internal/handler/arch_test.go @@ -101,21 +101,23 @@ func TestArchHandler_UnknownRepositoryReturns404(t *testing.T) { } } -// archUpstream is a fake pacman mirror. Files can be swapped and the mirror -// taken offline; requests are counted per path. +// archUpstream is a fake pacman mirror. Files can be swapped, single paths +// broken and the mirror taken offline; requests are counted per path. type archUpstream struct { *httptest.Server mu sync.Mutex files map[string][]byte modified map[string]time.Time hits map[string]int + broken map[string]bool + served func(path string) // called after a file is served, without mu available atomic.Bool auth atomic.Value } func newArchUpstream(t *testing.T, files map[string][]byte) *archUpstream { t.Helper() - u := &archUpstream{files: files, modified: map[string]time.Time{}, hits: map[string]int{}} + u := &archUpstream{files: files, modified: map[string]time.Time{}, hits: map[string]int{}, broken: map[string]bool{}} u.available.Store(true) u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !u.available.Load() { @@ -126,8 +128,14 @@ func newArchUpstream(t *testing.T, files map[string][]byte) *archUpstream { u.mu.Lock() data, ok := u.files[r.URL.Path] modified := u.modified[r.URL.Path] + broken := u.broken[r.URL.Path] + served := u.served u.hits[r.URL.Path]++ u.mu.Unlock() + if broken { + http.Error(w, "broken", http.StatusInternalServerError) + return + } if !ok { http.NotFound(w, r) return @@ -137,6 +145,9 @@ func newArchUpstream(t *testing.T, files map[string][]byte) *archUpstream { } w.Header().Set("Content-Type", "application/octet-stream") _, _ = w.Write(data) + if served != nil { + served(r.URL.Path) + } })) t.Cleanup(u.Close) return u @@ -206,10 +217,14 @@ func TestArchHandler_DatabasesServedUnchanged(t *testing.T) { t.Errorf("Authorization = %q, want %q", got, "Bearer arch-token") } + // core.db is fetched twice on its first request: once more after its + // signature, to make sure the two belong together. + wantHits := map[string]int{"/core/os/x86_64/core.db": 2} for _, path := range paths { assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux"+path, nil), path+" cached", files[path]) - if got := upstream.hitCount(path); got != 1 { - t.Errorf("%s: upstream hits = %d, want 1 (second read from cache)", path, got) + want := max(wantHits[path], 1) + if got := upstream.hitCount(path); got != want { + t.Errorf("%s: upstream hits = %d, want %d (second read from cache)", path, got, want) } } @@ -228,7 +243,7 @@ func TestArchHandler_StaleDatabaseServedWhenUpstreamDown(t *testing.T) { h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil), "seed", db) - ageMetadata(t, proxy, h, "archlinux", upstream.URL, "core/os/x86_64/core.db", 2*time.Hour) + ageMetadata(t, proxy, h, "archlinux", upstream.URL, "core/os/x86_64/core.db") upstream.available.Store(false) w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil) @@ -260,8 +275,8 @@ func TestArchHandler_DatabaseIfModifiedSince(t *testing.T) { // TestArchHandler_MissingDatabaseSignature covers the official repositories, // which sign no databases: pacman with DatabaseOptional asks for core.db.sig -// and must get a 404 while the upstream is up. With the upstream down the -// answer is a 502, which pacman also treats as "no signature". +// and must get a 404. The database refresh found it missing, so the answer +// needs no upstream and holds offline too. func TestArchHandler_MissingDatabaseSignature(t *testing.T) { upstream := newArchUpstream(t, map[string][]byte{"/core/os/x86_64/core.db": []byte("core-db-bytes")}) proxy := newArchTestProxy(t, upstream) @@ -271,11 +286,14 @@ func TestArchHandler_MissingDatabaseSignature(t *testing.T) { if w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db.sig", nil); w.Code != http.StatusNotFound { t.Errorf("missing signature: status = %d, want 404", w.Code) } + if got := upstream.hitCount("/core/os/x86_64/core.db.sig"); got != 1 { + t.Errorf("signature upstream hits = %d, want 1 (asked once, with the database)", got) + } upstream.available.Store(false) assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil), "db offline", []byte("core-db-bytes")) - if w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db.sig", nil); w.Code != http.StatusBadGateway { - t.Errorf("missing signature offline: status = %d, want 502", w.Code) + if w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db.sig", nil); w.Code != http.StatusNotFound { + t.Errorf("missing signature offline: status = %d, want 404", w.Code) } } @@ -302,7 +320,7 @@ func TestArchHandler_DatabaseSignatureFollowsDatabase(t *testing.T) { // the signature does. upstream.set(dbPath, []byte("db v2")) upstream.set(sigPath, []byte("sig v2")) - ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:], 2*time.Hour) + ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:]) assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v2", []byte("db v2")) assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig after db refresh", []byte("sig v2")) @@ -312,11 +330,11 @@ func TestArchHandler_DatabaseSignatureFollowsDatabase(t *testing.T) { } } -// TestArchHandler_OutdatedSignatureNeverServed covers a signature refetch that -// fails after the database was refreshed. The cached signature belongs to -// the old database, and pacman rejects a present but invalid signature even -// with DatabaseOptional, so the proxy must answer 404 or 502 rather than fall -// back to it. +// TestArchHandler_OutdatedSignatureNeverServed covers a signature fetch that +// fails while the database is refreshed. The cached signature belongs to the +// old database, and pacman rejects a present but invalid signature even with +// DatabaseOptional, so the proxy must answer 404 or 502 rather than fall back +// to it. func TestArchHandler_OutdatedSignatureNeverServed(t *testing.T) { const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" @@ -325,12 +343,8 @@ func TestArchHandler_OutdatedSignatureNeverServed(t *testing.T) { breakSig func(u *archUpstream) want int }{ - {"signature removed upstream", func(u *archUpstream) { - u.mu.Lock() - delete(u.files, sigPath) - u.mu.Unlock() - }, http.StatusNotFound}, - {"upstream down after database refresh", func(u *archUpstream) { u.available.Store(false) }, http.StatusBadGateway}, + {"signature removed upstream", func(u *archUpstream) { delete(u.files, sigPath) }, http.StatusNotFound}, + {"signature fetch fails", func(u *archUpstream) { u.broken[sigPath] = true }, http.StatusBadGateway}, } { t.Run(tt.name, func(t *testing.T) { upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) @@ -340,11 +354,13 @@ func TestArchHandler_OutdatedSignatureNeverServed(t *testing.T) { assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v1", []byte("db v1")) assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v1", []byte("sig v1")) - upstream.set(dbPath, []byte("db v2")) - ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:], 2*time.Hour) + upstream.mu.Lock() + upstream.files[dbPath] = []byte("db v2") + tt.breakSig(upstream) + upstream.mu.Unlock() + ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:]) assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v2", []byte("db v2")) - tt.breakSig(upstream) for i := range 2 { w := serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil) if w.Code != tt.want { @@ -355,6 +371,67 @@ func TestArchHandler_OutdatedSignatureNeverServed(t *testing.T) { } } +// TestArchHandler_DelayedSignatureMatchesCachedDatabase covers a signature +// first requested after the upstream moved on: it must be the one for the +// database the proxy still serves, not the upstream's new one. +func TestArchHandler_DelayedSignatureMatchesCachedDatabase(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v1", []byte("db v1")) + + upstream.set(dbPath, []byte("db v2")) + upstream.set(sigPath, []byte("sig v2")) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db within TTL", []byte("db v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "first signature request", []byte("sig v1")) +} + +// TestArchHandler_SignatureSurvivesUnchangedRefresh covers a database refresh +// that returns the same bytes: the cached signature still matches and must be +// served offline afterwards. +func TestArchHandler_SignatureSurvivesUnchangedRefresh(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db", []byte("db v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig", []byte("sig v1")) + + ageMetadata(t, proxy, h, "custom", upstream.URL, dbPath[1:]) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db refreshed unchanged", []byte("db v1")) + + upstream.available.Store(false) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig offline", []byte("sig v1")) + if got := upstream.hitCount(sigPath); got != 1 { + t.Errorf("signature upstream hits = %d, want 1", got) + } +} + +// TestArchHandler_DatabaseReplacedDuringRefresh covers a mirror that swaps +// database and signature between the proxy's two fetches. The proxy must +// notice and store the new pair, not the old database with the new signature. +func TestArchHandler_DatabaseReplacedDuringRefresh(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + var swapped atomic.Bool + upstream.served = func(path string) { + if path == sigPath && swapped.CompareAndSwap(false, true) { + upstream.set(dbPath, []byte("db v2")) + upstream.set(sigPath, []byte("sig v2")) + } + } + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db", []byte("db v2")) + upstream.available.Store(false) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig", []byte("sig v2")) +} + // TestArchHandler_RejectsQueryAndFragmentInPath covers a decoded '?' or '#' // in the path, which would otherwise fetch one file and cache it under the // name that follows. @@ -378,15 +455,14 @@ func TestArchHandler_RejectsQueryAndFragmentInPath(t *testing.T) { } } -// ageMetadata moves a cached metadata entry's fetch time into the past, as if -// it had been fetched age ago. -func ageMetadata(t *testing.T, proxy *Proxy, h *ArchHandler, repository, upstreamURL, path string, age time.Duration) { +// ageMetadata moves a metadata entry's fetch time past the one-hour TTL. +func ageMetadata(t *testing.T, proxy *Proxy, h *ArchHandler, repository, upstreamURL, path string) { t.Helper() entry, err := proxy.DB.GetMetadataCache(archEcosystem, h.metadataCacheKey(repository, upstreamURL, path)) if err != nil || entry == nil { t.Fatalf("metadata entry for %s: %v", path, err) } - entry.FetchedAt.Time = time.Now().Add(-age) + entry.FetchedAt.Time = time.Now().Add(-2 * time.Hour) if err := proxy.DB.UpsertMetadataCache(entry); err != nil { t.Fatalf("aging metadata entry: %v", err) } From 59f0203b00e28b162451dc8316cda474f9373ff1 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 17:08:17 +0200 Subject: [PATCH 6/7] When a signature has to be fetched on request because its fetch failed during the database refresh, it is now only served if it belongs to the database the proxy had cached at that point. If the upstream has published a new database in the meantime, the request fails with 502 instead of handing out a signature that cannot verify the database the client already has. The next sync gets the new pair. --- internal/handler/arch.go | 40 +++++++++++++++++++++---------- internal/handler/arch_test.go | 45 +++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 12 deletions(-) diff --git a/internal/handler/arch.go b/internal/handler/arch.go index f3505a42..64fde487 100644 --- a/internal/handler/arch.go +++ b/internal/handler/arch.go @@ -224,20 +224,27 @@ func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Req return } p.recordMetadataCacheMiss(archEcosystem) - // Nothing stored for this database yet, e.g. the signature fetch failed - // during the last refresh: refresh both. - res, _ := p.coalesceMetadata(r.Context(), d.coalesceKey("signature"), func(ctx context.Context) metadataResult { - if res, ok := h.cachedSignature(ctx, d); ok { + // Nothing stored for the cached database, e.g. its signature fetch failed + // during the last refresh. The client may already hold that database, so + // only a signature for it will do; if the refresh replaces it, fail. + entry, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) + served := h.databaseDigest(d) + res, _ := p.coalesceMetadata(r.Context(), d.coalesceKey("signature\x00"+served), func(ctx context.Context) metadataResult { + if res, ok := h.storedSignature(ctx, d, served); ok { return res } - entry, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) - if _, err := h.refresh(ctx, d, entry); err != nil { - return metadataResult{err: err} - } - if res, ok := h.cachedSignature(ctx, d); ok { + _, err := h.refresh(ctx, d, entry) + if served == "" { + if res, ok := h.cachedSignature(ctx, d); ok { + return res + } + } else if res, ok := h.storedSignature(ctx, d, served); ok { return res } - return metadataResult{err: errArchSignatureUnpaired} + if err == nil || h.databaseDigest(d) != served { + err = errArchSignatureUnpaired + } + return metadataResult{err: err} }) h.writeMetadata(w, r, d.sigKey, res) } @@ -292,11 +299,20 @@ func (h *ArchHandler) refresh(ctx context.Context, d archDatabase, entry *databa // cachedSignature returns the signature stored for the cached database. func (h *ArchHandler) cachedSignature(ctx context.Context, d archDatabase) (metadataResult, bool) { + dbDigest := h.databaseDigest(d) + if dbDigest == "" { + return metadataResult{}, false + } + return h.storedSignature(ctx, d, dbDigest) +} + +// databaseDigest returns the cached database's content digest, or "". +func (h *ArchHandler) databaseDigest(d archDatabase) string { current, _ := h.proxy.DB.GetMetadataCache(archEcosystem, d.dbKey) if current == nil || !current.ContentDigest.Valid { - return metadataResult{}, false + return "" } - return h.storedSignature(ctx, d, current.ContentDigest.String) + return current.ContentDigest.String } // storedSignature returns the signature stored for dbDigest. The storage diff --git a/internal/handler/arch_test.go b/internal/handler/arch_test.go index 1b71b3f7..2650f022 100644 --- a/internal/handler/arch_test.go +++ b/internal/handler/arch_test.go @@ -389,6 +389,51 @@ func TestArchHandler_DelayedSignatureMatchesCachedDatabase(t *testing.T) { assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "first signature request", []byte("sig v1")) } +// TestArchHandler_RecoveredSignatureMatchesServedDatabase covers a signature +// whose fetch failed while the database was served. If the upstream has moved +// on by the time pacman asks for the signature, the new one cannot verify the +// database pacman holds, so the proxy must fail rather than serve it. +func TestArchHandler_RecoveredSignatureMatchesServedDatabase(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + upstream.broken[sigPath] = true + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v1", []byte("db v1")) + + upstream.mu.Lock() + upstream.files[dbPath] = []byte("db v2") + upstream.files[sigPath] = []byte("sig v2") + upstream.broken[sigPath] = false + upstream.mu.Unlock() + + if w := serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil); w.Code != http.StatusBadGateway { + t.Errorf("signature for db v1: status = %d, body = %q, want 502", w.Code, w.Body.String()) + } + + // The next sync gets the new pair. + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v2", []byte("db v2")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v2", []byte("sig v2")) +} + +// TestArchHandler_RecoveredSignatureForUnchangedDatabase covers the same +// failed signature fetch when the database has not changed: the signature is +// fetched on request and served. +func TestArchHandler_RecoveredSignatureForUnchangedDatabase(t *testing.T) { + const dbPath, sigPath = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("db v1"), sigPath: []byte("sig v1")}) + upstream.broken[sigPath] = true + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+dbPath, nil), "db v1", []byte("db v1")) + upstream.mu.Lock() + upstream.broken[sigPath] = false + upstream.mu.Unlock() + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+sigPath, nil), "sig v1", []byte("sig v1")) +} + // TestArchHandler_SignatureSurvivesUnchangedRefresh covers a database refresh // that returns the same bytes: the cached signature still matches and must be // served offline afterwards. From 854182fa28e59ddebbe08ed56bc502fb5670c5d9 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 20:06:59 +0200 Subject: [PATCH 7/7] Concurrent clients could still end up with a database and a signature that don't belong together. Databases are now stored under their digest like signatures, so a cache row and its bytes always agree, also with several replicas on shared storage. Writes for one database are serialized, and the signature recovery shares the refresh with database requests, so a slow fetch can no longer store an older database over a newer one. When the cache write fails, the proxy serves the database the cache still describes. Unchanged databases are no longer rewritten, and a remembered missing signature is asked for again after the metadata TTL. The new tests force these interleavings with storage and upstream hooks. --- internal/handler/arch.go | 156 +++++++++-- internal/handler/arch_concurrency_test.go | 315 ++++++++++++++++++++++ 2 files changed, 445 insertions(+), 26 deletions(-) create mode 100644 internal/handler/arch_concurrency_test.go diff --git a/internal/handler/arch.go b/internal/handler/arch.go index 64fde487..6706dae4 100644 --- a/internal/handler/arch.go +++ b/internal/handler/arch.go @@ -3,11 +3,15 @@ package handler import ( "context" "crypto/sha256" + "database/sql" "encoding/hex" "errors" + "fmt" "net/http" "regexp" "strings" + "sync" + "time" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/metrics" @@ -32,6 +36,7 @@ type ArchHandler struct { proxy *Proxy proxyURL string repositories map[string]string + locks sync.Map // database cache key -> *sync.Mutex } // NewArchHandler creates an Arch Linux handler. Without repositories it serves @@ -198,14 +203,7 @@ func (h *ArchHandler) handleDatabase(w http.ResponseWriter, r *http.Request, rep return } p.recordMetadataCacheMiss(archEcosystem) - res, _ := p.coalesceMetadata(r.Context(), d.coalesceKey("database"), func(ctx context.Context) metadataResult { - entry, hit := p.cachedMetadataState(ctx, archEcosystem, d.dbKey, nil) - if hit != nil { - return *hit - } - res, _ := h.refresh(ctx, d, entry) - return res - }) + res := h.sharedRefresh(r.Context(), d, true) h.writeMetadata(w, r, d.dbKey, res) } @@ -227,13 +225,14 @@ func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Req // Nothing stored for the cached database, e.g. its signature fetch failed // during the last refresh. The client may already hold that database, so // only a signature for it will do; if the refresh replaces it, fail. - entry, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) served := h.databaseDigest(d) res, _ := p.coalesceMetadata(r.Context(), d.coalesceKey("signature\x00"+served), func(ctx context.Context) metadataResult { if res, ok := h.storedSignature(ctx, d, served); ok { return res } - _, err := h.refresh(ctx, d, entry) + if res := h.sharedRefresh(ctx, d, false); res.err != nil { + return res + } if served == "" { if res, ok := h.cachedSignature(ctx, d); ok { return res @@ -241,34 +240,46 @@ func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Req } else if res, ok := h.storedSignature(ctx, d, served); ok { return res } - if err == nil || h.databaseDigest(d) != served { - err = errArchSignatureUnpaired - } - return metadataResult{err: err} + return metadataResult{err: errArchSignatureUnpaired} }) h.writeMetadata(w, r, d.sigKey, res) } var errArchSignatureUnpaired = errors.New("database signature could not be paired with the database") +// sharedRefresh runs one refresh per database at a time, so a slow fetch +// cannot store an older database over one cached after it started. With +// ifStale, a refresh that finds the database fresh again skips the upstream. +func (h *ArchHandler) sharedRefresh(ctx context.Context, d archDatabase, ifStale bool) metadataResult { + p := h.proxy + res, _ := p.coalesceMetadata(ctx, d.coalesceKey("database"), func(ctx context.Context) metadataResult { + entry, hit := p.cachedMetadataState(ctx, archEcosystem, d.dbKey, nil) + if hit != nil && ifStale { + return *hit + } + res, _ := h.refresh(ctx, d, entry) + return res + }) + return res +} + // archPairAttempts bounds refetches while the mirror replaces the database. const archPairAttempts = 3 // refresh fetches the database, then its signature, and stores the signature -// for the database's content. It returns the database and why no signature -// could be stored, if so. +// for the database's content. It returns the database as cached and why no +// signature could be stored, if so. func (h *ArchHandler) refresh(ctx context.Context, d archDatabase, entry *database.MetadataCacheEntry) (metadataResult, error) { p := h.proxy - res := p.fetchMetadataFromUpstream(ctx, archEcosystem, d.dbKey, d.dbURL, archAccept, "identity", nil, entry) + res, row := h.fetchDatabase(ctx, d, entry) if res.err != nil { return res, res.err } for range archPairAttempts { - current, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) - if current == nil || !current.ContentDigest.Valid { + if row == nil || !row.ContentDigest.Valid { return res, errArchSignatureUnpaired } - dbDigest := current.ContentDigest.String + dbDigest := row.ContentDigest.String if _, ok := h.storedSignature(ctx, d, dbDigest); ok { return res, nil } @@ -283,7 +294,7 @@ func (h *ArchHandler) refresh(ctx context.Context, d archDatabase, entry *databa } // The mirror may have replaced the database in the meantime. - again, err := p.fetchUpstreamMetadata(ctx, d.dbURL, current, archAccept, "identity") + again, err := p.fetchUpstreamMetadata(ctx, d.dbURL, row, archAccept, "identity") if err != nil { return res, err } @@ -291,12 +302,92 @@ func (h *ArchHandler) refresh(ctx context.Context, d archDatabase, entry *databa h.storeSignature(ctx, d, dbDigest, sig) return res, nil } - p.cacheMetadataBlob(ctx, archEcosystem, d.dbKey, metadataStoragePath(archEcosystem, d.dbKey), again) - res = metadataResult{body: again.body, contentType: again.contentType, contentEncoding: again.contentEncoding} + if res, row = h.storeDatabase(ctx, d, again); res.err != nil { + return res, res.err + } } return res, errArchSignatureUnpaired } +// fetchDatabase fetches and caches the database, falling back to the cached +// copy when the upstream fails. +func (h *ArchHandler) fetchDatabase(ctx context.Context, d archDatabase, entry *database.MetadataCacheEntry) (metadataResult, *database.MetadataCacheEntry) { + p := h.proxy + meta, err := p.fetchUpstreamMetadata(ctx, d.dbURL, entry, archAccept, "identity") + if errors.Is(err, errStale304) { + meta, err = p.fetchUpstreamMetadata(ctx, d.dbURL, nil, archAccept, "identity") + } + if err == nil { + return h.storeDatabase(ctx, d, meta) + } + if entry == nil { + return metadataResult{err: fmt.Errorf("upstream failed and no cached metadata: %w", err)}, nil + } + p.Logger.Warn("upstream metadata fetch failed, checking cache", "ecosystem", archEcosystem, "key", d.dbKey, "error", err) + return h.readDatabase(ctx, p.currentMetadataEntry(archEcosystem, d.dbKey, entry), err) +} + +// storeDatabase caches meta under its digest and returns the database the +// cache row now describes. That is meta unless the write failed; bytes the +// row does not describe would be paired with the wrong signature. +func (h *ArchHandler) storeDatabase(ctx context.Context, d archDatabase, meta *upstreamMetadata) (metadataResult, *database.MetadataCacheEntry) { + p := h.proxy + path := archBlobPath(d.dbKey, archDigest(meta.body)) + defer h.lock(d.dbKey)() + + previous, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) + if previous != nil && previous.StoragePath == path { + // Same bytes: a failed rewrite would delete the blob the row uses. + h.touchDatabase(previous, meta) + } else { + p.cacheMetadataBlob(ctx, archEcosystem, d.dbKey, path, meta) + } + row, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) + if row == nil { + // Nothing cached at all: serve it like an uncached proxy would. + return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding}, nil + } + if row.StoragePath != path { + return h.readDatabase(ctx, row, errArchDatabaseNotCached) + } + if previous != nil && previous.StoragePath != path { + _ = p.Storage.Delete(ctx, previous.StoragePath) + } + return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding}, row +} + +// touchDatabase marks an unchanged cached database as fetched now. +func (h *ArchHandler) touchDatabase(row *database.MetadataCacheEntry, meta *upstreamMetadata) { + updated := *row + updated.ETag = sql.NullString{String: meta.etag, Valid: meta.etag != ""} + updated.LastModified = sql.NullTime{Time: meta.lastModified, Valid: !meta.lastModified.IsZero()} + updated.FetchedAt = sql.NullTime{Time: time.Now(), Valid: true} + if err := h.proxy.DB.UpsertMetadataCache(&updated); err != nil { + h.proxy.Logger.Warn("failed to record cached metadata", "ecosystem", archEcosystem, "key", row.Name, "error", err) + } +} + +var errArchDatabaseNotCached = errors.New("database could not be cached") + +// readDatabase returns the cached database row's bytes, or cause if they are +// unreadable. +func (h *ArchHandler) readDatabase(ctx context.Context, row *database.MetadataCacheEntry, cause error) (metadataResult, *database.MetadataCacheEntry) { + body, contentType, err := h.proxy.readCachedMetadata(ctx, row, nil) + if err != nil { + return metadataResult{err: fmt.Errorf("cached database unusable (%v): %w", err, cause)}, nil + } + return metadataResult{body: body, contentType: contentType, contentEncoding: row.ContentEncoding.String}, row +} + +// lock serializes cache writes for one database and its signature, so a row +// never points at a blob another write has just removed. +func (h *ArchHandler) lock(dbKey string) func() { + m, _ := h.locks.LoadOrStore(dbKey, &sync.Mutex{}) + mu := m.(*sync.Mutex) + mu.Lock() + return mu.Unlock +} + // cachedSignature returns the signature stored for the cached database. func (h *ArchHandler) cachedSignature(ctx context.Context, d archDatabase) (metadataResult, bool) { dbDigest := h.databaseDigest(d) @@ -318,15 +409,20 @@ func (h *ArchHandler) databaseDigest(d archDatabase) string { // storedSignature returns the signature stored for dbDigest. The storage // path names the digest, so a refresh with unchanged bytes keeps it. func (h *ArchHandler) storedSignature(ctx context.Context, d archDatabase, dbDigest string) (metadataResult, bool) { - entry, _ := h.proxy.DB.GetMetadataCache(archEcosystem, d.sigKey) + p := h.proxy + entry, _ := p.DB.GetMetadataCache(archEcosystem, d.sigKey) if entry == nil { return metadataResult{}, false } switch entry.StoragePath { case archMissingSignaturePath(d, dbDigest): + // Ask again after the TTL, in case the 404 was transient. + if p.MetadataTTL <= 0 || !entry.FetchedAt.Valid || time.Since(entry.FetchedAt.Time) >= p.MetadataTTL { + return metadataResult{}, false + } return metadataResult{err: ErrUpstreamNotFound}, true case archSignaturePath(d, dbDigest): - body, contentType, err := h.proxy.readCachedMetadata(ctx, entry, nil) + body, contentType, err := p.readCachedMetadata(ctx, entry, nil) if err != nil { return metadataResult{}, false } @@ -344,6 +440,8 @@ func (h *ArchHandler) storeSignature(ctx context.Context, d archDatabase, dbDige path = archMissingSignaturePath(d, dbDigest) sig = &upstreamMetadata{contentType: archPackageMediaType} } + defer h.lock(d.dbKey)() + previous, _ := p.DB.GetMetadataCache(archEcosystem, d.sigKey) p.cacheMetadataBlob(ctx, archEcosystem, d.sigKey, path, sig) if previous == nil || previous.StoragePath == path { @@ -354,8 +452,14 @@ func (h *ArchHandler) storeSignature(ctx context.Context, d archDatabase, dbDige } } +// archBlobPath names a database or signature blob after the database digest +// it belongs to, so a cache row and its bytes always agree. +func archBlobPath(key, dbDigest string) string { + return "_metadata/" + archEcosystem + "/" + key + "/" + strings.TrimPrefix(dbDigest, "sha256:") +} + func archSignaturePath(d archDatabase, dbDigest string) string { - return "_metadata/" + archEcosystem + "/" + d.sigKey + "/" + strings.TrimPrefix(dbDigest, "sha256:") + return archBlobPath(d.sigKey, dbDigest) } func archMissingSignaturePath(d archDatabase, dbDigest string) string { diff --git a/internal/handler/arch_concurrency_test.go b/internal/handler/arch_concurrency_test.go new file mode 100644 index 00000000..a2f374e1 --- /dev/null +++ b/internal/handler/arch_concurrency_test.go @@ -0,0 +1,315 @@ +package handler + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/storage" +) + +// archHookStorage wraps a storage with hooks to force interleavings. +type archHookStorage struct { + storage.Storage + failStore func(path string) error + afterStore func(path string) + beforeDelete func(path string) +} + +func (s *archHookStorage) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { + if s.failStore != nil { + if err := s.failStore(path); err != nil { + return 0, "", err + } + } + n, hash, err := s.Storage.Store(ctx, path, r) + if s.afterStore != nil { + s.afterStore(path) + } + return n, hash, err +} + +func (s *archHookStorage) Delete(ctx context.Context, path string) error { + if s.beforeDelete != nil { + s.beforeDelete(path) + } + return s.Storage.Delete(ctx, path) +} + +const archTestDB, archTestSig = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig" + +// pauseOnce returns a hook that stops the first call for a path starting with +// prefix until release +// closes or a short timeout passes, so a test does not hang when locking +// keeps the other writer out. +func pauseOnce(prefix string, paused, release chan struct{}) func(string) { + var fired atomic.Bool + return func(p string) { + if !strings.HasPrefix(p, prefix) || !fired.CompareAndSwap(false, true) { + return + } + close(paused) + select { + case <-release: + case <-time.After(200 * time.Millisecond): + } + } +} + +// waitPaused waits until a pauseOnce hook fired. +func waitPaused(t *testing.T, paused <-chan struct{}) { + t.Helper() + select { + case <-paused: + case <-time.After(5 * time.Second): + t.Fatal("hook never fired") + } +} + +// waitBriefly waits for done, or gives up after a moment when a lock holds the +// call back. +func waitBriefly(done <-chan struct{}) { + select { + case <-done: + case <-time.After(300 * time.Millisecond): + } +} + +// TestArchHandler_ReplicasWritingOutOfOrder covers two proxy replicas sharing +// database and storage, whose database writes finish out of order: the cache +// row must never describe other bytes than its blob, or every client gets the +// newest database with an older signature. +func TestArchHandler_ReplicasWritingOutOfOrder(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1"), archTestSig: []byte("sig v1")}) + upstream.broken[archTestSig] = true + proxy := newArchTestProxy(t, upstream) + replicaA := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + replicaB := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + d := replicaA.database("custom", upstream.URL, archTestDB[1:]) + + assertArchBody(t, serveArchRequest(replicaA, http.MethodGet, "/custom"+archTestDB, nil), "db v1", []byte("db v1")) + upstream.mu.Lock() + upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v2"), []byte("sig v2") + upstream.broken[archTestSig] = false + upstream.mu.Unlock() + ageMetadata(t, proxy, replicaA, "custom", upstream.URL, archTestDB[1:]) + + // Replica A's signature recovery stores db v2 and stalls before its row + // update. + paused, release := make(chan struct{}), make(chan struct{}) + proxy.Storage = &archHookStorage{Storage: proxy.Storage, + afterStore: pauseOnce("_metadata/"+archEcosystem+"/"+d.dbKey+"/", paused, release)} + recovered := make(chan struct{}) + go func() { + serveArchRequest(replicaA, http.MethodGet, "/custom"+archTestSig, nil) + close(recovered) + }() + waitPaused(t, paused) + + // Replica B pairs v2, then caches v3 whose signature fetch fails. + assertArchBody(t, serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestDB, nil), "B db v2", []byte("db v2")) + assertArchBody(t, serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestSig, nil), "B sig v2", []byte("sig v2")) + upstream.mu.Lock() + upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v3"), []byte("sig v3") + upstream.broken[archTestSig] = true + upstream.mu.Unlock() + ageMetadata(t, proxy, replicaB, "custom", upstream.URL, archTestDB[1:]) + assertArchBody(t, serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestDB, nil), "B db v3", []byte("db v3")) + + close(release) + <-recovered + + db := serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestDB, nil) + sig := serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestSig, nil) + if sig.Code == http.StatusOK && strings.Replace(db.Body.String(), "db", "sig", 1) != sig.Body.String() { + t.Errorf("%q served with %q", db.Body.String(), sig.Body.String()) + } +} + +// TestArchHandler_FailedDatabaseWriteKeepsPair covers a database refresh whose +// cache write fails: the proxy must not hand out the new database while the +// cache still pairs the old one with its signature. +func TestArchHandler_FailedDatabaseWriteKeepsPair(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1"), archTestSig: []byte("sig v1")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + d := h.database("custom", upstream.URL, archTestDB[1:]) + var fail atomic.Bool + proxy.Storage = &archHookStorage{Storage: proxy.Storage, failStore: func(path string) error { + if fail.Load() && strings.HasPrefix(path, archBlobPath(d.dbKey, "")) { + return errors.New("disk full") + } + return nil + }} + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db v1", []byte("db v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil), "sig v1", []byte("sig v1")) + upstream.set(archTestDB, []byte("db v2")) + upstream.set(archTestSig, []byte("sig v2")) + ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:]) + fail.Store(true) + + db := serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil) + sig := serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil) + if db.Code == http.StatusOK && sig.Code == http.StatusOK && strings.Replace(db.Body.String(), "db", "sig", 1) != sig.Body.String() { + t.Errorf("%q served with %q", db.Body.String(), sig.Body.String()) + } +} + +// TestArchHandler_ConcurrentSignatureWritesLeaveOneBlob covers two signature +// writes overlapping: afterwards exactly one signature blob is left and the +// cache row points at it. +func TestArchHandler_ConcurrentSignatureWritesLeaveOneBlob(t *testing.T) { + d1, d2, d3 := archDigest([]byte("1")), archDigest([]byte("2")), archDigest([]byte("3")) + + for _, tt := range []struct { + name string + hook func(h *ArchHandler, d archDatabase, s *archHookStorage, paused, release chan struct{}) + first string // digest stored by the stalled write + other string // digest stored by the write overtaking it + }{ + {"stalled after store", func(_ *ArchHandler, d archDatabase, s *archHookStorage, paused, release chan struct{}) { + s.afterStore = pauseOnce(archSignaturePath(d, d2), paused, release) + }, d2, d3}, + {"stalled before delete", func(_ *ArchHandler, d archDatabase, s *archHookStorage, paused, release chan struct{}) { + s.beforeDelete = pauseOnce(archSignaturePath(d, d1), paused, release) + }, d2, d1}, + } { + t.Run(tt.name, func(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{}) + proxy := newArchTestProxy(t, upstream) + mock := proxy.Storage.(*mockStorage) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + d := h.database("custom", upstream.URL, archTestDB[1:]) + ctx := context.Background() + h.storeSignature(ctx, d, d1, &upstreamMetadata{body: []byte("sig 1")}) + + hooks := &archHookStorage{Storage: mock} + proxy.Storage = hooks + paused, release := make(chan struct{}), make(chan struct{}) + tt.hook(h, d, hooks, paused, release) + + stalled := make(chan struct{}) + go func() { + h.storeSignature(ctx, d, tt.first, &upstreamMetadata{body: []byte("sig first")}) + close(stalled) + }() + waitPaused(t, paused) + overtaking := make(chan struct{}) + go func() { + h.storeSignature(ctx, d, tt.other, &upstreamMetadata{body: []byte("sig other")}) + close(overtaking) + }() + waitBriefly(overtaking) + close(release) + <-stalled + <-overtaking + + row, _ := proxy.DB.GetMetadataCache(archEcosystem, d.sigKey) + if _, err := mock.Open(ctx, row.StoragePath); err != nil { + t.Errorf("row points at %s, which is gone: %v", row.StoragePath, err) + } + mock.mu.Lock() + defer mock.mu.Unlock() + var blobs []string + for path := range mock.files { + if strings.HasPrefix(path, archBlobPath(d.sigKey, "")) { + blobs = append(blobs, path) + } + } + if len(blobs) != 1 { + t.Errorf("signature blobs = %v, want only the row's", blobs) + } + }) + } +} + +// TestArchHandler_MissingSignatureRecheckedAfterTTL covers a 404 for the +// signature that was only transient: once the TTL is over the proxy asks +// again instead of answering 404 until the next database is published. +func TestArchHandler_MissingSignatureRecheckedAfterTTL(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1")}) + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db", []byte("db v1")) + if w := serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil); w.Code != http.StatusNotFound { + t.Fatalf("signature: status = %d, want 404", w.Code) + } + + upstream.set(archTestSig, []byte("sig v1")) + ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:]) + ageMetadata(t, proxy, h, "custom", upstream.URL, archTestSig[1:]) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db refreshed", []byte("db v1")) + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil), "signature after TTL", []byte("sig v1")) +} + +// TestArchHandler_SlowFetchDoesNotOverwriteNewerDatabase covers a signature +// recovery whose database fetch is still in flight while other clients +// refresh to newer databases. Its result must not replace the newer cached +// database, or a client holding that one gets an older signature. +func TestArchHandler_SlowFetchDoesNotOverwriteNewerDatabase(t *testing.T) { + upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1"), archTestSig: []byte("sig v1")}) + upstream.broken[archTestSig] = true + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db v1", []byte("db v1")) + upstream.mu.Lock() + upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v2"), []byte("sig v2") + upstream.broken[archTestSig] = false + upstream.mu.Unlock() + ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:]) + + // Client A's signature recovery has fetched db v2; the response is slow. + paused, release := make(chan struct{}), make(chan struct{}) + upstream.mu.Lock() + upstream.served = pauseOnce(archTestDB, paused, release) + upstream.mu.Unlock() + recovered := make(chan struct{}) + go func() { + serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil) + close(recovered) + }() + waitPaused(t, paused) + + // Client B refreshes meanwhile, then the upstream moves to v3, whose + // signature fails, and client C refreshes to it. + b := make(chan struct{}) + go func() { + serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil) + serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil) + close(b) + }() + waitBriefly(b) + upstream.mu.Lock() + upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v3"), []byte("sig v3") + upstream.broken[archTestSig] = true + upstream.mu.Unlock() + ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:]) + var cDB string + c := make(chan struct{}) + go func() { + cDB = serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil).Body.String() + close(c) + }() + waitBriefly(c) + + close(release) + <-recovered + <-b + <-c + + if sig := serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil); sig.Code == http.StatusOK && strings.Replace(cDB, "db", "sig", 1) != sig.Body.String() { + t.Errorf("client holding %q got %q", cDB, sig.Body.String()) + } + if got := serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil).Body.String(); got < cDB { + t.Errorf("cache went back to %q after serving %q", got, cDB) + } +}