diff --git a/docs/common/caching.md b/docs/common/caching.md index b49b16c..66cc8e8 100644 --- a/docs/common/caching.md +++ b/docs/common/caching.md @@ -16,8 +16,6 @@ Every proxied request on a remote is classified as **denied**, **immutable** or | `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, and never for immutable paths. | | `stale_on_error` | On an upstream network error, serve the stored copy instead of failing. | | `serve_stale` | rpm/deb/alpine virtual members only. When a virtual re-merges, use this member's previous index at once and refresh it in the background; the next merge picks up the refresh. The first merge still waits, and so does any merge once the previous index is over an hour old, so a member that keeps failing surfaces its error. Default `false`. | - -`PUT` replaces the whole remote: any field omitted from the body, such as `stale_on_error` or `serve_stale`, resets to its default (`mutable_ttl` is the exception above). | `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. | diff --git a/internal/virtual/engine.go b/internal/virtual/engine.go index 21225b6..07cf58c 100644 --- a/internal/virtual/engine.go +++ b/internal/virtual/engine.go @@ -337,6 +337,7 @@ func (e *Engine) merged(ctx context.Context, key string, merge func(context.Cont e.mu.Unlock() v, err, _ := e.sf.Do(key, func() (any, error) { + e.pruneStale() repo, err := merge(context.WithoutCancel(ctx)) if err != nil { return nil, err @@ -556,26 +557,29 @@ func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch return fetch(ctx) } e.mu.Lock() - for k, se := range e.stale { - if time.Since(se.at) >= maxStaleAge { - delete(e.stale, k) - } + if e.stale == nil { + e.stale = map[string]*staleEntry{} } prev := e.stale[sk] + // A cold fetch holds a placeholder so eviction or toggle-off during it is seen. + token := prev + if prev == nil || prev.m == nil { + prev, token = nil, &staleEntry{at: time.Now()} + e.stale[sk] = token + } e.mu.Unlock() refresh := func() (any, error) { m, err := fetch(ctx) - if err == nil { - e.mu.Lock() - // An evicted or replaced entry means this refresh is no longer wanted. - if e.stale[sk] == prev { - if e.stale == nil { - e.stale = map[string]*staleEntry{} - } + e.mu.Lock() + // An evicted or replaced entry means this refresh is no longer wanted. + if e.stale[sk] == token { + if err == nil { e.stale[sk] = &staleEntry{m: m, at: time.Now()} + } else if prev == nil { + delete(e.stale, sk) } - e.mu.Unlock() } + e.mu.Unlock() return m, err } if prev == nil { @@ -591,6 +595,18 @@ func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch return prev.m.(*T), nil } +// pruneStale drops serve_stale copies past maxStaleAge, including those of +// removed members and virtuals. +func (e *Engine) pruneStale() { + e.mu.Lock() + defer e.mu.Unlock() + for k, se := range e.stale { + if time.Since(se.at) >= maxStaleAge { + delete(e.stale, k) + } + } +} + // evictLRU drops the least recently used merges under prefix beyond keep. func (e *Engine) evictLRU(prefix string, keep int) { e.mu.Lock() diff --git a/internal/virtual/engine_stale_test.go b/internal/virtual/engine_stale_test.go index 62e4c0c..2b409c7 100644 --- a/internal/virtual/engine_stale_test.go +++ b/internal/virtual/engine_stale_test.go @@ -271,3 +271,60 @@ func TestServeStaleEvictionDropsLateRefresh(t *testing.T) { t.Fatalf("stale holds %d copies, want %d", len(e.stale), maxDebSuites) } } + +func TestServeStaleEvictionDropsLateColdFetch(t *testing.T) { + release := make(chan struct{}) + var slow atomic.Bool + e := fakeEngine(nil) + e.mergeTTL = time.Minute + e.getRemote = staleRemote + e.debMember = func(_ context.Context, name, base string) (*DebMember, error) { + if base == "dists/s0/" && slow.Load() { + <-release + } + return suiteDebMember(name, "1"), nil + } + fetch := func(suite string) { + if _, _, err := e.Fetch(context.Background(), debVirt("a"), "dists/"+suite+"/Release", ""); err != nil { + t.Error(err) + } + } + sk := "deb\x00v\x00dists/s0/\x00a" + fetch("s0") + slow.Store(true) + e.mu.Lock() + e.gens["deb\x00v\x00dists/s0/"].at = time.Time{} + delete(e.stale, sk) + e.mu.Unlock() + done := make(chan struct{}) + go func() { defer close(done); fetch("s0") }() + for { + e.mu.Lock() + cold := e.stale[sk] != nil + e.mu.Unlock() + if cold { + break + } + time.Sleep(time.Millisecond) + } + for i := range maxDebSuites { + time.Sleep(time.Microsecond) + fetch(fmt.Sprintf("s%d", i+1)) + } + close(release) + <-done + e.mu.Lock() + defer e.mu.Unlock() + if e.stale[sk] != nil { + t.Fatal("a cold fetch finishing after eviction must not re-insert its copy") + } +} + +func TestMergePrunesExpiredCopiesOfOtherMembers(t *testing.T) { + e, _ := staleEngine(false, func() *RPMMember { return rpmPkgMember("a", "1") }) + e.stale = map[string]*staleEntry{"rpm\x00gone\x00x": {m: rpmPkgMember("x", "1"), at: time.Now().Add(-maxStaleAge)}} + fetchRepomd(t, e) + if len(e.stale) != 0 { + t.Fatal("any merge should prune expired copies of removed members") + } +}