Merge master into benvin/remote-query-forwarder

2026-10-18 03:15:17 +11:00
parent 769d673a4a d88a047788
commit 71e6458c77
185 changed files with 8550 additions and 1091 deletions
@@ -134,10 +134,11 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 		slog.Warn("cache hit but S3 miss, re-fetching", "remote", remote.Name, "path", path)
 	}

-	locked, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL)
+	token, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL)
 	if err != nil {
 		slog.Warn("lock acquire failed", "error", err)
 	}
+	locked := token != ""

 	if !locked {
 		// Another request holds the fetch lock. Poll the store until the leader
@@ -151,7 +152,7 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 	}

 	if locked {
-		defer e.cache.ReleaseLock(ctx, remote.Name, path)
+		defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remote.Name, path, token) }()
 	}

 	if class == ClassMutable && remote.CheckMutable {
@@ -160,7 +161,7 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 			notModified, err := e.checkUpstream(ctx, remote, path, etag, prov)
 			if err == nil && notModified {
 				_ = e.cache.SetTTL(ctx, remote.Name, path, ttl)
-				_ = e.cache.SetETag(ctx, remote.Name, path, etag, ttl)
+				_ = e.cache.SetETag(ctx, remote.Name, path, etag, etagTTL(ttl))
 				result, err := e.serveFromStore(ctx, remote, path)
 				if err == nil {
 					result.Source = "cache"
@@ -229,10 +230,11 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
 		return fmt.Errorf("get remote: %w", err)
 	}
 	if !wildcard {
-		if err := e.waitForLock(ctx, remoteName, path); err != nil {
+		release, err := e.Lock(ctx, remoteName, path, e.evictLockWait, fetchLockTTL)
+		if err != nil {
 			return err
 		}
-		defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }()
+		defer release()
 		if err := e.db.DeleteArtifact(ctx, remoteName, path); err != nil {
 			return fmt.Errorf("delete artifact: %w", err)
 		}
@@ -265,28 +267,36 @@ func (e *Engine) evictPrefix(ctx context.Context, remoteName, prefix string) err
 	return e.cache.ForgetPrefix(ctx, remoteName, prefix)
 }

-// waitForLock takes the per-path fetch lock so an in-flight Fetch cannot
-// re-set TTL/ETag keys after an evict. It fails with a 503 when the lock
-// cannot be taken within evictLockWait or Redis errors.
-func (e *Engine) waitForLock(ctx context.Context, remoteName, path string) error {
-	deadline := time.Now().Add(e.evictLockWait)
+// Lock takes the per-path Redis lock (the one Fetch holds) and returns its
+// release. It fails with a 503 when the lock cannot be taken within wait or
+// Redis errors.
+func (e *Engine) Lock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (func(), error) {
+	token, err := e.waitForLock(ctx, remoteName, path, wait, ttl)
+	if err != nil {
+		return nil, err
+	}
+	return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path, token) }, nil
+}
+
+func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (string, error) {
+	deadline := time.Now().Add(wait)
 	for {
-		ok, err := e.cache.AcquireLock(ctx, remoteName, path, fetchLockTTL)
-		if ok {
-			return nil
+		token, err := e.cache.AcquireLock(ctx, remoteName, path, ttl)
+		if token != "" {
+			return token, nil
 		}
 		if ctx.Err() != nil {
-			return ctx.Err()
+			return "", ctx.Err()
 		}
 		if err != nil {
-			return &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("fetch lock: %v", err)}
+			return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("lock: %v", err)}
 		}
 		if time.Now().After(deadline) {
-			return &ProxyError{Status: http.StatusServiceUnavailable, Message: "fetch in progress, retry evict"}
+			return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: "resource busy, retry later"}
 		}
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return "", ctx.Err()
 		case <-time.After(50 * time.Millisecond):
 		}
 	}
@@ -366,7 +376,7 @@ func (e *Engine) headUpstreamOnce(ctx context.Context, remote models.Remote, pat
 				req.Header.Set(k, v)
 			}
 		}
-		return http.DefaultClient.Do(req)
+		return clientForRemote(remote).Do(req)
 	}

 	resp, err := doHead(nil)
@@ -504,8 +514,9 @@ func (e *Engine) fetchFromUpstreamOnce(ctx context.Context, remote models.Remote
 		}

 		_ = e.cache.SetTTL(ctx, remote.Name, path, ttl)
-		if etag := resp.Header.Get("ETag"); etag != "" {
-			_ = e.cache.SetETag(ctx, remote.Name, path, etag, ttl)
+		// A zero TTL never expires, so it never needs revalidating.
+		if remote.CheckMutable && ttl > 0 {
+			_ = e.cache.SetETag(ctx, remote.Name, path, resp.Header.Get("ETag"), etagTTL(ttl))
 		}

 		return &FetchResult{
@@ -533,9 +544,6 @@ func (e *Engine) fetchFromUpstreamOnce(ctx context.Context, remote models.Remote
 	}

 	_ = e.cache.SetTTL(ctx, remote.Name, path, ttl)
-	if etag := resp.Header.Get("ETag"); etag != "" {
-		_ = e.cache.SetETag(ctx, remote.Name, path, etag, ttl)
-	}

 	reader, info, err := e.store.Download(ctx, casResult.S3Key)
 	if err != nil {
@@ -655,6 +663,12 @@ func (e *Engine) checkUpstreamOnce(ctx context.Context, remote models.Remote, pa
 	return resp.StatusCode == http.StatusNotModified, nil
 }

+// etagTTL keeps a mutable path's ETag well past its freshness key so an expired
+// path revalidates with a conditional HEAD, while idle paths still age out.
+func etagTTL(ttl time.Duration) time.Duration {
+	return max(10*ttl, 7*24*time.Hour)
+}
+
 func (e *Engine) ttlFor(remote models.Remote, class Classification) time.Duration {
 	switch class {
 	case ClassImmutable:
@@ -17,12 +17,14 @@ import (
 	"git.unkin.net/unkin/artifactapi/internal/storage"
 	"git.unkin.net/unkin/artifactapi/internal/testsupport"
 	"git.unkin.net/unkin/artifactapi/pkg/models"
+	goredis "github.com/redis/go-redis/v9"
 )

 var (
 	testEngine *Engine
 	testCache  *cache.Redis
 	testDB     *database.DB
+	testRedis  string
 	upstream   *httptest.Server
 )

@@ -64,6 +66,7 @@ func TestMain(m *testing.M) {
 	}

 	testCache = redis
+	testRedis = redisURL
 	testDB = db
 	testEngine = NewEngine(db, redis, s3)
 	upstream = httptest.NewServer(http.HandlerFunc(mockUpstream))
@@ -216,6 +219,20 @@ func TestHead(t *testing.T) {
 	}
 }

+func TestHeadUpstreamUsesRemoteTimeout(t *testing.T) {
+	slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(3 * time.Second)
+	}))
+	defer slow.Close()
+	r := models.Remote{Name: "slow-head", PackageType: models.PackageGeneric, BaseURL: slow.URL, UpstreamResponseHeaderTimeout: 1}
+
+	start := time.Now()
+	_, err := (&Engine{}).headUpstreamOnce(context.Background(), r, "blob.bin", prov(t, models.PackageGeneric))
+	if err == nil || time.Since(start) > 2*time.Second {
+		t.Fatalf("HEAD ignored the remote's response header timeout: err=%v after %s", err, time.Since(start))
+	}
+}
+
 func TestStaleOnError(t *testing.T) {
 	requireStack(t)
 	ctx := context.Background()
@@ -260,30 +277,169 @@ func TestCircuitOpenServesStale(t *testing.T) {
 	res.Reader.Close()
 }

+// expire drops path's freshness key so the next fetch revalidates.
+func expire(t *testing.T, remote, path string) {
+	t.Helper()
+	if err := testCache.SetTTL(context.Background(), remote, path, time.Millisecond); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(5 * time.Millisecond)
+}
+
+func mutableFetch(t *testing.T, r models.Remote, path string) (string, string) {
+	t.Helper()
+	res, err := testEngine.Fetch(context.Background(), r, path, prov(t, models.PackageRPM))
+	if err != nil {
+		t.Fatalf("fetch: %v", err)
+	}
+	return readAll(t, res), res.Source
+}
+
+const repomd = "8/Everything/x86_64/repodata/repomd.xml"
+
 func TestMutableRevalidation(t *testing.T) {
+	requireStack(t)
+	srv, rev, conditional := changingUpstream(t)
+	r := seed(t, models.Remote{Name: "eng-reval", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: srv.URL, CheckMutable: true, MutableTTL: 60})
+
+	if body, _ := mutableFetch(t, r, repomd); body != "rev1:/"+repomd {
+		t.Fatalf("initial fetch = %q", body)
+	}
+	expire(t, r.Name, repomd)
+	if body, src := mutableFetch(t, r, repomd); body != "rev1:/"+repomd || src != "cache" {
+		t.Errorf("after expiry = %q from %s, want cached rev1", body, src)
+	}
+	if n := conditional.Load(); n != 1 {
+		t.Errorf("conditional requests = %d, want 1 revalidation", n)
+	}
+	if fresh, _ := testCache.CheckTTL(context.Background(), r.Name, repomd); !fresh {
+		t.Error("304 should re-arm the freshness key")
+	}
+
+	rev.Store("rev2")
+	expire(t, r.Name, repomd)
+	if body, src := mutableFetch(t, r, repomd); body != "rev2:/"+repomd || src != "remote" {
+		t.Errorf("after upstream change = %q from %s, want remote rev2", body, src)
+	}
+}
+
+func etagKeyTTL(t *testing.T, remote, path string) time.Duration {
+	t.Helper()
+	opts, err := goredis.ParseURL(testRedis)
+	if err != nil {
+		t.Fatal(err)
+	}
+	c := goredis.NewClient(opts)
+	defer func() { _ = c.Close() }()
+	d, err := c.PTTL(context.Background(), "etag:"+remote+":"+path).Result()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return d
+}
+
+// The ETag is only kept when it can be used, and never without an expiry.
+func TestMutableETagStorage(t *testing.T) {
+	requireStack(t)
+	srv, _, _ := changingUpstream(t)
+	for _, tc := range []struct {
+		name  string
+		check bool
+		ttl   int
+		want  time.Duration
+	}{
+		{"eng-etag-off", false, 60, -2},
+		{"eng-etag-never", true, 0, -2},
+		{"eng-etag-short", true, 60, 7 * 24 * time.Hour},
+		{"eng-etag-long", true, 86400, 10 * 24 * time.Hour},
+	} {
+		r := seed(t, models.Remote{Name: tc.name, PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: srv.URL, CheckMutable: tc.check, MutableTTL: tc.ttl})
+		mutableFetch(t, r, repomd)
+		got := etagKeyTTL(t, tc.name, repomd)
+		if tc.want < 0 {
+			if got != tc.want {
+				t.Errorf("%s: etag key ttl = %v, want no key", tc.name, got)
+			}
+			continue
+		}
+		if got <= tc.want-time.Minute || got > tc.want {
+			t.Errorf("%s: etag key ttl = %v, want ~%v", tc.name, got, tc.want)
+		}
+	}
+}
+
+// Immutable paths are never revalidated, so their upstream ETag is not kept.
+func TestImmutableFetchStoresNoETag(t *testing.T) {
+	requireStack(t)
+	srv, _, _ := changingUpstream(t)
+	r := seed(t, models.Remote{Name: "eng-etag-imm", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: srv.URL, CheckMutable: true, MutableTTL: 60})
+	const rpm = "Packages/e2e-1.0-1.noarch.rpm"
+	if body, src := mutableFetch(t, r, rpm); body != "rev1:/"+rpm || src != "remote" {
+		t.Fatalf("fetch = %q from %s", body, src)
+	}
+	if got := etagKeyTTL(t, r.Name, rpm); got != -2 {
+		t.Errorf("etag key ttl = %v, want no key", got)
+	}
+}
+
+// A 304 for a path whose S3 body is gone falls back to a full refetch.
+func TestNotModifiedWithEvictedBodyRefetches(t *testing.T) {
+	requireStack(t)
+	srv, _, conditional := changingUpstream(t)
+	r := seed(t, models.Remote{Name: "eng-304-gone", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: srv.URL, CheckMutable: true, MutableTTL: 60})
+
+	mutableFetch(t, r, repomd)
+	if err := testEngine.store.Delete(context.Background(), storage.IndexKey(r.Name, repomd)); err != nil {
+		t.Fatal(err)
+	}
+	expire(t, r.Name, repomd)
+	if body, src := mutableFetch(t, r, repomd); body != "rev1:/"+repomd || src != "remote" {
+		t.Errorf("got %q from %s, want refetched rev1 from remote", body, src)
+	}
+	if n := conditional.Load(); n != 1 {
+		t.Errorf("conditional requests = %d, want the 304 HEAD first", n)
+	}
+}
+
+// A failing revalidation HEAD still serves the stored copy with stale_on_error.
+func TestStaleOnErrorWhenRevalidationFails(t *testing.T) {
+	requireStack(t)
+	srv, _, _ := changingUpstream(t)
+	r := seed(t, models.Remote{Name: "eng-stale-head", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: srv.URL, CheckMutable: true, MutableTTL: 60, StaleOnError: true})
+
+	mutableFetch(t, r, repomd)
+	srv.Close()
+	expire(t, r.Name, repomd)
+	if body, src := mutableFetch(t, r, repomd); body != "rev1:/"+repomd || src != "cache" {
+		t.Errorf("got %q from %s, want stale rev1 from cache", body, src)
+	}
+}
+
+// After a remote is deleted and flushed, a re-created remote of the same name
+// fetches from its own upstream instead of revalidating the old ETag.
+func TestRecreatedRemoteHasNoStaleState(t *testing.T) {
 	requireStack(t)
 	ctx := context.Background()
-	r := seed(t, models.Remote{Name: "eng-npm", PackageType: models.PackageNPM, RepoType: models.RepoTypeRemote, BaseURL: upstream.URL, CheckMutable: true, MutableTTL: 3600, StaleOnError: true})
-	p := prov(t, models.PackageNPM)
+	oldSrv, _, _ := changingUpstream(t)
+	r := seed(t, models.Remote{Name: "eng-recreate", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: oldSrv.URL, CheckMutable: true, MutableTTL: 60})
+	mutableFetch(t, r, repomd)

-	res, err := testEngine.Fetch(ctx, r, "pkg", p)
-	if err != nil {
-		t.Fatalf("initial mutable fetch: %v", err)
+	if err := testDB.DeleteRemote(ctx, r.Name); err != nil {
+		t.Fatal(err)
 	}
-	res.Reader.Close()
-
-	// Expire only the freshness marker; the ETag persists, forcing a
-	// conditional revalidation that the upstream answers with 304.
-	testCache.SetTTL(ctx, "eng-npm", "pkg", time.Millisecond)
-	time.Sleep(10 * time.Millisecond)
-	res, err = testEngine.Fetch(ctx, r, "pkg", p)
-	if err != nil {
-		t.Fatalf("revalidation fetch: %v", err)
+	if err := testCache.FlushRemote(ctx, r.Name); err != nil {
+		t.Fatal(err)
 	}
-	if res.Source != "cache" {
-		t.Errorf("revalidated response should come from cache, got %s", res.Source)
+	newSrv, rev, conditional := changingUpstream(t)
+	rev.Store("rev9")
+	r.BaseURL = newSrv.URL
+	seed(t, r)
+	if body, src := mutableFetch(t, r, repomd); body != "rev9:/"+repomd || src != "remote" {
+		t.Errorf("got %q from %s, want rev9 from the new upstream", body, src)
+	}
+	if n := conditional.Load(); n != 0 {
+		t.Errorf("conditional requests = %d, want none after flush", n)
 	}
-	res.Reader.Close()
 }

 func TestBearerTokenFlow(t *testing.T) {
@@ -59,7 +59,7 @@ type cached struct{ row, index, ttl, etag bool }
 // Mutable indexes live in the S3 index; immutable blobs get an artifact row.
 var (
 	indexCached = cached{index: true, ttl: true, etag: true}
-	blobCached  = cached{row: true, ttl: true, etag: true}
+	blobCached  = cached{row: true, ttl: true}
 )

 func layers(t *testing.T, remote, path string) cached {
@@ -152,13 +152,15 @@ func TestEvictUnknownRemote(t *testing.T) {
 	}
 }

-func holdLock(t *testing.T, remote, path string) {
+func holdLock(t *testing.T, remote, path string) string {
 	t.Helper()
 	ctx := context.Background()
-	if ok, err := testCache.AcquireLock(ctx, remote, path, time.Minute); !ok || err != nil {
-		t.Fatalf("acquire: %v %v", ok, err)
+	token, err := testCache.AcquireLock(ctx, remote, path, time.Minute)
+	if token == "" || err != nil {
+		t.Fatalf("acquire: %q %v", token, err)
 	}
-	t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path) })
+	t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path, token) })
+	return token
 }

 func TestEvictWaitsForFetchLock(t *testing.T) {
@@ -167,7 +169,7 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
 	r := rpmRemote(t, "evict-lock", srv.URL)
 	ctx := context.Background()
 	const path = "8/Everything/x86_64/repodata/repomd.xml"
-	holdLock(t, r.Name, path)
+	token := holdLock(t, r.Name, path)
 	done := make(chan error, 1)
 	go func() { done <- testEngine.Evict(ctx, r.Name, path) }()
 	select {
@@ -175,13 +177,13 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
 		t.Fatalf("evict returned while fetch lock held: %v", err)
 	case <-time.After(200 * time.Millisecond):
 	}
-	_ = testCache.ReleaseLock(ctx, r.Name, path)
+	_ = testCache.ReleaseLock(ctx, r.Name, path, token)
 	if err := <-done; err != nil {
 		t.Fatalf("evict: %v", err)
 	}
-	ok, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
-	if !ok || err != nil {
-		t.Fatalf("lock still held after evict: %v %v", ok, err)
+	token, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
+	if token == "" || err != nil {
+		t.Fatalf("lock still held after evict: %q %v", token, err)
 	}
 }