From 3c76902533b410b1ab6dcec7123e773819bfcb7d Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 17 Oct 2026 16:51:37 +1100 Subject: [PATCH] Bound the mutable ETag TTL and flush Redis state on remote delete --- docs/common/caching.md | 3 +- internal/api/v2/remotes.go | 5 + internal/api/v2/remotes_flush_test.go | 25 +++++ internal/proxy/engine.go | 13 ++- internal/proxy/engine_test.go | 151 +++++++++++++++++++++++--- 5 files changed, 178 insertions(+), 19 deletions(-) diff --git a/docs/common/caching.md b/docs/common/caching.md index f4ea8c8..977377e 100644 --- a/docs/common/caching.md +++ b/docs/common/caching.md @@ -13,7 +13,7 @@ Every proxied request on a remote is classified as **denied**, **immutable** or | `mutable_patterns` | Regexes forcing a path mutable. | | `immutable_ttl` | Seconds an immutable path stays fresh. `0` = forever. | | `mutable_ttl` | Seconds a mutable path stays fresh. `0` also means never expires. Omitted on create, it defaults to `3600`; omitted on update, the stored value is kept. | -| `check_mutable` | When a mutable path's `mutable_ttl` expires, sends a conditional `HEAD` (`If-None-Match`) with the stored upstream ETag. A `304` re-arms the TTL and serves the stored copy; anything else refetches in full. The ETag has no expiry of its own and is replaced or dropped on each full fetch. | +| `check_mutable` | When a mutable path's `mutable_ttl` expires, sends a conditional `HEAD` (`If-None-Match`) with the stored upstream ETag. A `304` re-arms the TTL and serves the stored copy; anything else refetches in full. The ETag is kept for `max(10 × mutable_ttl, 7 days)`, re-armed on each `304`, so idle paths age out. It is only stored when `check_mutable` is set and `mutable_ttl` is non-zero. | | `stale_on_error` | On an upstream network error, serve the stored copy instead of failing. | | `quarantine_enabled` / `quarantine_days` | Stored on the remote but not enforced by the proxy yet. | | `mirrorlist` / `mirror_strategy` | rpm/deb/alpine only. Upstream pool is `[base_url] + mirrorlist`; `round_robin` (default) rotates the starting mirror, `least_conn` starts at the one with the fewest in-flight requests. Network errors and 5xx fail over to the next mirror; other statuses return as-is. | @@ -60,6 +60,7 @@ curl -X DELETE https://artifactapi.example.com/api/v2/remotes//objects//*` and takes no lock. - Returns 204, or 404 for an unknown remote. diff --git a/internal/api/v2/remotes.go b/internal/api/v2/remotes.go index 3aac4bc..b5c3a82 100644 --- a/internal/api/v2/remotes.go +++ b/internal/api/v2/remotes.go @@ -170,6 +170,11 @@ func (h *RemotesHandler) del(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusInternalServerError) return } + if h.cache != nil { + if err := h.cache.FlushRemote(r.Context(), name); err != nil { + slog.Warn("flush cached metadata after delete failed", "remote", name, "error", err) + } + } w.WriteHeader(http.StatusNoContent) } diff --git a/internal/api/v2/remotes_flush_test.go b/internal/api/v2/remotes_flush_test.go index 09a25b5..4f8036a 100644 --- a/internal/api/v2/remotes_flush_test.go +++ b/internal/api/v2/remotes_flush_test.go @@ -94,3 +94,28 @@ func TestUpdateFlushFailureStillSucceeds(t *testing.T) { t.Fatalf("flush calls = %v, want exactly one attempted flush", ff.calls) } } + +// Deleting a remote flushes its cached state so a re-created remote of the +// same name starts cold; a flush error does not fail the delete. +func TestDeleteFlushesCache(t *testing.T) { + if testDSN == "" { + t.Skip("Docker unavailable") + } + db, err := database.New(testDSN) + if err != nil { + t.Fatal(err) + } + defer db.Close() + + for _, ff := range []*fakeFlusher{{}, {err: errors.New("redis down")}} { + const name = "rpm-flush-delete" + seedRemote(t, db, name, "https://example.com/repo") + h := NewRemotesHandler(db, ff, nil).Routes() + if c := do(t, h, "DELETE", "/"+name, ""); c != 204 { + t.Fatalf("delete = %d, want 204", c) + } + if len(ff.calls) != 1 || ff.calls[0] != name { + t.Fatalf("flush calls = %v, want exactly one flush of %q", ff.calls, name) + } + } +} diff --git a/internal/proxy/engine.go b/internal/proxy/engine.go index 68cd841..ed0efc6 100644 --- a/internal/proxy/engine.go +++ b/internal/proxy/engine.go @@ -160,6 +160,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, etagTTL(ttl)) result, err := e.serveFromStore(ctx, remote, path) if err == nil { result.Source = "cache" @@ -490,8 +491,10 @@ func (e *Engine) fetchFromUpstreamOnce(ctx context.Context, remote models.Remote } _ = e.cache.SetTTL(ctx, remote.Name, path, ttl) - // The ETag outlives the freshness key so an expired path is revalidated. - _ = e.cache.SetETag(ctx, remote.Name, path, resp.Header.Get("ETag"), 0) + // 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{ Reader: io.NopCloser(bytesReader(body)), @@ -640,6 +643,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: diff --git a/internal/proxy/engine_test.go b/internal/proxy/engine_test.go index 9678f9f..e366d2e 100644 --- a/internal/proxy/engine_test.go +++ b/internal/proxy/engine_test.go @@ -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)) @@ -260,41 +263,157 @@ 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: 1}) - const path = "8/Everything/x86_64/repodata/repomd.xml" - fetch := func() (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 - } + r := seed(t, models.Remote{Name: "eng-reval", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: srv.URL, CheckMutable: true, MutableTTL: 60}) - if body, _ := fetch(); body != "rev1:/"+path { + if body, _ := mutableFetch(t, r, repomd); body != "rev1:/"+repomd { t.Fatalf("initial fetch = %q", body) } - time.Sleep(1100 * time.Millisecond) - if body, src := fetch(); body != "rev1:/"+path || src != "cache" { + 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, path); !fresh { + if fresh, _ := testCache.CheckTTL(context.Background(), r.Name, repomd); !fresh { t.Error("304 should re-arm the freshness key") } rev.Store("rev2") - time.Sleep(1100 * time.Millisecond) - if body, src := fetch(); body != "rev2:/"+path || src != "remote" { + 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) + } + } +} + +// 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() + 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) + + if err := testDB.DeleteRemote(ctx, r.Name); err != nil { + t.Fatal(err) + } + if err := testCache.FlushRemote(ctx, r.Name); err != nil { + t.Fatal(err) + } + 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) + } +} + func TestBearerTokenFlow(t *testing.T) { requireStack(t) ctx := context.Background()