Enable storage retention for conda

docs/configuration.md

@@ -92,7 +92,7 @@ A package rule wins over an ecosystem rule, which wins over `default`. Package r
 | `cargo` | yes | `pkg:cargo/serde`, the crate name as published; case, `-` and `_` have to match, so `pkg:cargo/inflector` doesn't cover `Inflector`. |
 | `composer` | yes | `pkg:composer/symfony/console`; case doesn't matter. A key without a vendor is refused. |
 | `conan` | yes | `pkg:conan/zlib`, the recipe name; a rule covers every version, revision and user/channel of it. |
-| `conda` | no | - |
+| `conda` | yes | `pkg:conda/conda-forge/numpy`, with the channel as the namespace; the package name's case doesn't matter. A key without a channel is refused. |
 | `cran` | yes | `pkg:cran/ggplot2`, the package name as published, case included (`pkg:cran/Rcpp`). A rule covers source and binary builds alike. |
 | `deb` | no | - |
 | `gem` | no | - |

internal/handler/conda.go

@@ -5,15 +5,49 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
+	"slices"
 	"strings"
 	"time"
+
+	"github.com/git-pkgs/proxy/internal/retention"
+	"github.com/git-pkgs/purl"
 )
 
 const (
 	condaUpstream = "https://conda.anaconda.org"
 	minCondaParts = 3 // name-version-build requires at least 3 hyphen-separated parts
 )
 
+// Packages are stored under the name {channel}/{name}, which purl writes as
+// pkg:conda/{channel}%2F{name}. Package rules name the channel as the
+// namespace, pkg:conda/conda-forge/numpy, and are mapped to that form.
+func init() {
+	retention.Default.Register(retention.Spec{Key: "conda", CanonicalPackage: condaRetentionKey})
+}
+
+// condaRetentionKey maps a package rule to the stored package PURL. It
+// takes the channel from the namespace, or from the name in the stored
+// form. Downloads are routed as /{channel}/{arch}/{filename}, so the channel
+// is always one path segment; a key without a channel, with more segments
+// or with an empty one is refused, since no stored package looks like that.
+// The package name is lowercased the way conda treats it; the channel stays
+// as the URL had it.
+func condaRetentionKey(p *purl.PURL) (string, bool) {
+	if p == nil {
+		return "", false
+	}
+	full := p.Name
+	if p.Namespace != "" {
+		full = p.Namespace + "/" + p.Name
+	}
+	segments := strings.Split(full, "/")
+	if len(segments) != 2 || slices.Contains(segments, "") {
+		return "", false
+	}
+	channel, name := segments[0], strings.ToLower(segments[1])
+	return retention.DefaultCanonical("conda")(purl.New(p.Type, "", channel+"/"+name, "", nil))
+}
+
 // CondaHandler handles Conda/Anaconda registry protocol requests.
 type CondaHandler struct {
 	proxy       *Proxy

internal/handler/conda_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/git-pkgs/cooldown"
+	"github.com/git-pkgs/purl"
 )
 
 func TestCondaParseFilename(t *testing.T) {
@@ -303,3 +304,35 @@ func TestCondaHandleRepodataWithoutCooldown(t *testing.T) {
 		t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
 	}
 }
+
+func TestCondaRetentionKey(t *testing.T) {
+	tests := []struct {
+		key  string
+		want string
+		ok   bool
+	}{
+		{"pkg:conda/conda-forge/numpy", "pkg:conda/conda-forge%2Fnumpy", true},
+		{"pkg:conda/conda-forge%2Fnumpy", "pkg:conda/conda-forge%2Fnumpy", true},
+		{"pkg:conda/conda-forge/PyYAML", "pkg:conda/conda-forge%2Fpyyaml", true},
+		{"pkg:conda/Conda-Forge/numpy", "pkg:conda/Conda-Forge%2Fnumpy", true},
+		{"pkg:conda/conda-forge/label/rc/numpy", "", false},
+		{"pkg:conda/conda-forge%2Flabel%2Frc%2Fnumpy", "", false},
+		{"pkg:conda/numpy", "", false},
+		{"pkg:conda/%2Fnumpy", "", false},
+		{"pkg:conda/conda-forge%2F", "", false},
+		{"pkg:npm/conda-forge/numpy", "", false},
+	}
+	for _, tt := range tests {
+		p, err := purl.Parse(tt.key)
+		if err != nil {
+			t.Fatalf("Parse(%s): %v", tt.key, err)
+		}
+		got, ok := condaRetentionKey(p)
+		if ok != tt.ok || (ok && got != tt.want) {
+			t.Errorf("condaRetentionKey(%s) = %q, %v, want %q, %v", tt.key, got, ok, tt.want, tt.ok)
+		}
+	}
+	if _, ok := condaRetentionKey(nil); ok {
+		t.Error("condaRetentionKey(nil) accepted")
+	}
+}

internal/server/retention_conda_test.go

@@ -0,0 +1,201 @@
+package server
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/git-pkgs/proxy/internal/config"
+	"github.com/git-pkgs/proxy/internal/database"
+	"github.com/git-pkgs/proxy/internal/handler"
+	"github.com/git-pkgs/proxy/internal/retention"
+	"github.com/git-pkgs/registries/fetch"
+)
+
+const (
+	condaForgeNumpyPath = "/conda-forge/linux-64/numpy-1.24.0-py311h64a7726_0.conda"
+	condaAnacondaNumpy  = "/anaconda/linux-64/numpy-1.24.0-py311h08b1b3b_0.conda"
+	condaRequestsPath   = "/conda-forge/noarch/requests-2.31.0-pyhd8ed1ab_0.conda"
+)
+
+// condaRetentionUpstream serves package files the way conda.anaconda.org
+// does, counting downloads.
+func condaRetentionUpstream(t *testing.T) (*httptest.Server, *atomic.Int64) {
+	t.Helper()
+	var downloads atomic.Int64
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case condaForgeNumpyPath, condaAnacondaNumpy, condaRequestsPath:
+			downloads.Add(1)
+			_, _ = io.WriteString(w, "conda package bytes")
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	t.Cleanup(srv.Close)
+	return srv, &downloads
+}
+
+func newCondaRetentionProxy(t *testing.T, upstream *httptest.Server) (*httptest.Server, *database.DB) {
+	t.Helper()
+	db, store := setupEvictionTest(t)
+	// The upstream is a plain-HTTP loopback server; the default fetcher
+	// refuses loopback, a stock client reaches it.
+	fetcher := fetch.NewFetcher(fetch.WithHTTPClient(upstream.Client()), fetch.WithMaxRetries(0))
+	t.Cleanup(func() { _ = fetcher.Close() })
+	proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), discardLogger())
+	proxy.HTTPClient = upstream.Client()
+	h := handler.NewCondaHandlerWithUpstream(proxy, "http://proxy.test", upstream.URL)
+	srv := httptest.NewServer(h.Routes())
+	t.Cleanup(srv.Close)
+	return srv, db
+}
+
+func condaGet(t *testing.T, url string) {
+	t.Helper()
+	resp, err := http.Get(url)
+	if err != nil {
+		t.Fatalf("GET %s: %v", url, err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("GET %s = %d %s, want 200", url, resp.StatusCode, body)
+	}
+}
+
+// condaArtifact returns the record of the package file named filename.
+func condaArtifact(t *testing.T, db *database.DB, filename string) database.Artifact {
+	t.Helper()
+	var art database.Artifact
+	if err := db.Get(&art, db.Rebind(`SELECT * FROM artifacts WHERE filename = ?`), filename); err != nil {
+		t.Fatalf("reading artifact %s: %v", filename, err)
+	}
+	return art
+}
+
+func condaBackdate(t *testing.T, db *database.DB, filename string, fetched, accessed time.Time) {
+	t.Helper()
+	query := db.Rebind(`UPDATE artifacts SET fetched_at = ?, last_accessed_at = ? WHERE filename = ?`)
+	if _, err := db.Exec(query, fetched, accessed, filename); err != nil {
+		t.Fatalf("backdating %s: %v", filename, err)
+	}
+}
+
+func TestCondaRegistersForRetention(t *testing.T) {
+	if _, ok := retention.Default.Lookup("conda"); !ok {
+		t.Fatal("conda is not registered for retention")
+	}
+	if key, ok := retention.Default.KeyForDBEcosystem("conda"); !ok || key != "conda" {
+		t.Errorf("KeyForDBEcosystem(conda) = %q, %v, want conda, true", key, ok)
+	}
+}
+
+// A package file served from the cache stays while it is used, is evicted
+// once its rule runs out, comes back from upstream on the next download,
+// and stays after that even with the evicted copy's old access time on the
+// record.
+func TestCondaRetentionEvictsAndRefetches(t *testing.T) {
+	upstream, downloads := condaRetentionUpstream(t)
+	srv, db := newCondaRetentionProxy(t, upstream)
+	const file = "requests-2.31.0-pyhd8ed1ab_0.conda"
+
+	condaGet(t, srv.URL+condaRequestsPath)
+	condaGet(t, srv.URL+condaRequestsPath)
+	if got := downloads.Load(); got != 1 {
+		t.Fatalf("upstream downloads = %d, want 1", got)
+	}
+	if !condaArtifact(t, db, file).LastAccessedAt.Valid {
+		t.Fatal("cache hit did not record an access")
+	}
+
+	rules := retention.Rules{Ecosystems: map[string]time.Duration{"conda": retentionDay}}
+	now := time.Now()
+	sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now)
+	if !condaArtifact(t, db, file).StoragePath.Valid {
+		t.Fatal("package used today was evicted")
+	}
+
+	old := now.Add(-2 * retentionDay)
+	condaBackdate(t, db, file, old, old)
+	sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now)
+	if condaArtifact(t, db, file).StoragePath.Valid {
+		t.Fatal("package unused for two days kept under a one-day rule")
+	}
+
+	condaGet(t, srv.URL+condaRequestsPath)
+	if got := downloads.Load(); got != 2 {
+		t.Errorf("upstream downloads = %d, want 2 after the eviction", got)
+	}
+	art := condaArtifact(t, db, file)
+	condaBackdate(t, db, file, art.FetchedAt.Time, now.Add(-30*retentionDay))
+	sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, time.Now())
+	if !condaArtifact(t, db, file).StoragePath.Valid {
+		t.Error("refetched package evicted again by the next sweep")
+	}
+}
+
+// A key without a channel parses as a conda PURL but can never match a
+// stored package, so it is refused at validation. A key with the channel
+// as the namespace is accepted.
+func TestCondaRetentionRefusesKeysWithoutChannel(t *testing.T) {
+	cfg := config.Default()
+	cfg.Storage.Retention.Packages = map[string]string{"pkg:conda/conda-forge/numpy": "0"}
+	if _, err := cfg.RetentionRules(retention.Default); err != nil {
+		t.Fatalf("RetentionRules(pkg:conda/conda-forge/numpy): %v", err)
+	}
+	cfg.Storage.Retention.Packages = map[string]string{"pkg:conda/numpy": "0"}
+	_, err := cfg.RetentionRules(retention.Default)
+	if err == nil || !strings.Contains(err.Error(), "does not match how conda packages are stored") {
+		t.Errorf("RetentionRules(pkg:conda/numpy) error = %v, want the key refused", err)
+	}
+}
+
+// A package rule with the channel as the namespace reaches the package in
+// that channel only; the same package from another channel and other
+// packages fall under the conda rule.
+func TestCondaRetentionPackageRule(t *testing.T) {
+	upstream, _ := condaRetentionUpstream(t)
+	srv, db := newCondaRetentionProxy(t, upstream)
+	const forgeNumpy, anacondaNumpy, requests = "numpy-1.24.0-py311h64a7726_0.conda", "numpy-1.24.0-py311h08b1b3b_0.conda", "requests-2.31.0-pyhd8ed1ab_0.conda"
+	condaGet(t, srv.URL+condaForgeNumpyPath)
+	condaGet(t, srv.URL+condaAnacondaNumpy)
+	condaGet(t, srv.URL+condaRequestsPath)
+
+	var stored string
+	art := condaArtifact(t, db, forgeNumpy)
+	if err := db.Get(&stored, db.Rebind(`SELECT package_purl FROM versions WHERE purl = ?`), art.VersionPURL); err != nil {
+		t.Fatalf("reading stored package PURL: %v", err)
+	}
+	if stored != "pkg:conda/conda-forge%2Fnumpy" {
+		t.Errorf("stored package PURL = %q, want pkg:conda/conda-forge%%2Fnumpy", stored)
+	}
+
+	old := time.Now().Add(-30 * retentionDay)
+	for _, filename := range []string{forgeNumpy, anacondaNumpy, requests} {
+		condaBackdate(t, db, filename, old, old)
+	}
+	cfg := config.Default()
+	cfg.Storage.Retention.Ecosystems = map[string]string{"conda": "1d"}
+	cfg.Storage.Retention.Packages = map[string]string{"pkg:conda/conda-forge/numpy": "0"}
+	rules, err := cfg.RetentionRules(retention.Default)
+	if err != nil {
+		t.Fatalf("RetentionRules: %v", err)
+	}
+	sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, time.Now())
+
+	if !condaArtifact(t, db, forgeNumpy).StoragePath.Valid {
+		t.Error("conda-forge numpy evicted despite its never-evict rule")
+	}
+	if condaArtifact(t, db, anacondaNumpy).StoragePath.Valid {
+		t.Error("anaconda numpy kept by the rule for conda-forge")
+	}
+	if condaArtifact(t, db, requests).StoragePath.Valid {
+		t.Error("requests kept past the conda rule")
+	}
+}