Guard serve_stale cold fetches against eviction and prune copies on every merge

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. |
 

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()

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")
+	}
+}