feat(cache): heal divergent tier 0 on ranged reads

internal/cache/tiered.go

@@ -25,6 +25,10 @@ type Tiered struct {
 	metadata  *metadatadb.Store
 	etags     *metadatadb.Map[Key, string]
 	namespace Namespace
+	// healer refreshes a stale tier 0 from a deeper tier when a ranged pinned
+	// read is served from the deeper tier (ranged bodies can't backfill). Shared
+	// across namespaced copies via the pointer. Nil for a single-tier cache.
+	healer *tieredHealer
 }
 
 // MaybeNewTiered creates a [Tiered] cache from one or more caches.
@@ -41,7 +45,7 @@ func MaybeNewTiered(ctx context.Context, caches []Cache, metadata *metadatadb.St
 	if metadata == nil {
 		panic("Tiered cache requires a metadata store")
 	}
-	return Tiered{caches: caches, metadata: metadata, etags: tieredETags(metadata, "")}
+	return Tiered{caches: caches, metadata: metadata, etags: tieredETags(metadata, ""), healer: newTieredHealer(ctx)}
 }
 
 type authoritativeCache struct {
@@ -60,6 +64,12 @@ var _ Cache = (*Tiered)(nil)
 
 // Close all underlying caches.
 func (t Tiered) Close() error {
+	// Drain in-flight heals before closing caches so they never operate on a
+	// closed backend.
+	if t.healer != nil {
+		t.healer.cancel()
+		t.healer.wg.Wait()
+	}
 	wg := sync.WaitGroup{}
 	errs := make([]error, len(t.caches))
 	for i, cache := range t.caches {
@@ -276,10 +286,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose
 		if fallback != nil {
 			discardTieredReader(ctx, key, fallback)
 		}
-		if i > 0 && !partial {
-			r = t.backfillReader(ctx, key, r, headers, t.caches[0])
-		}
-		return r, headers, nil
+		return t.convergeTier0(ctx, key, r, headers, c, i, partial), headers, nil
 	}
 	if len(probeErrs) > 0 {
 		if fallback != nil {
@@ -365,6 +372,166 @@ func backfillCreateOptions(headers http.Header) []Option {
 	return []Option{WithETag(rawETag)}
 }
 
+const (
+	// tieredHealMaxConcurrency bounds concurrent full-object refreshes so a
+	// popular stale object can't stampede the shared tier.
+	tieredHealMaxConcurrency = 8
+	// tieredHealTimeout bounds a single refresh copy.
+	tieredHealTimeout = 5 * time.Minute
+)
+
+var errHealSuperseded = errors.New("heal superseded by concurrent write")
+
+// tieredHealer performs asynchronous, bounded, deduplicated full-object
+// refreshes of tier 0 from a deeper tier.
+type tieredHealer struct {
+	mu       sync.Mutex
+	inflight map[string]struct{}
+	sem      chan struct{}
+	ctx      context.Context
+	cancel   context.CancelFunc
+	wg       sync.WaitGroup
+}
+
+func newTieredHealer(ctx context.Context) *tieredHealer {
+	// Detach from the constructor's cancellation but keep its values, then add
+	// our own cancel so Close can drain in-flight heals.
+	hctx, cancel := context.WithCancel(context.WithoutCancel(ctx)) //nolint:gosec // cancel is stored and called in Close
+	return &tieredHealer{
+		inflight: map[string]struct{}{},
+		sem:      make(chan struct{}, tieredHealMaxConcurrency),
+		ctx:      hctx,
+		cancel:   cancel,
+	}
+}
+
+// convergeTier0 makes tier 0 catch up to a body served from a deeper tier: a
+// full body backfills tier 0 as the caller reads it; a ranged body (which must
+// never be stored as the whole object) instead schedules an async full-object
+// refresh (#368). A body served from tier 0 needs no convergence.
+func (t Tiered) convergeTier0(ctx context.Context, key Key, r io.ReadCloser, headers http.Header, source Cache, tier int, partial bool) io.ReadCloser {
+	switch {
+	case tier == 0:
+		return r
+	case !partial:
+		return t.backfillReader(ctx, key, r, headers, t.caches[0])
+	default:
+		t.healTier0(ctx, key, source, headers.Get(ETagKey))
+		return r
+	}
+}
+
+// healTier0 schedules a refresh of tier 0 from source, gated on the served
+// version being the authoritative one tracked for this key. If tier 0 is
+// legitimately newer than a lagging deeper tier, the tracked etag names tier
+// 0's version rather than the served one, so the heal is skipped — a newer
+// tier 0 is never overwritten with an older deeper version.
+func (t Tiered) healTier0(ctx context.Context, key Key, source Cache, servedETag string) {
+	if t.healer == nil || servedETag == "" {
+		return
+	}
+	want, ok := t.etags.Get(key)
+	if !ok || want != servedETag {
+		return
+	}
+	t.healer.trigger(ctx, t.namespace, key, source, t.caches[0], want, t.etags)
+}
+
+func (h *tieredHealer) trigger(
+	reqCtx context.Context,
+	namespace Namespace,
+	key Key,
+	source, dst Cache,
+	wantETag string,
+	etags *metadatadb.Map[Key, string],
+) {
+	// Namespace-scope the dedup key so heals in different namespaces don't
+	// suppress each other.
+	dedupKey := string(namespace) + "\x00" + string(key[:])
+
+	h.mu.Lock()
+	if _, ok := h.inflight[dedupKey]; ok {
+		h.mu.Unlock()
+		return
+	}
+	// Opportunistic: skip if the refresh pool is saturated. A later divergent
+	// read re-triggers, so no heal is permanently lost.
+	select {
+	case h.sem <- struct{}{}:
+	default:
+		h.mu.Unlock()
+		return
+	}
+	h.inflight[dedupKey] = struct{}{}
+	h.mu.Unlock()
+
+	// Detach from the request's lifetime but keep its logger; use the healer's
+	// context so Close can cancel in-flight heals.
+	ctx := logging.ContextWithLogger(h.ctx, logging.FromContext(reqCtx))
+	h.wg.Go(func() {
+		defer func() {
+			h.mu.Lock()
+			delete(h.inflight, dedupKey)
+			h.mu.Unlock()
+			<-h.sem
+		}()
+		h.run(ctx, key, source, dst, wantETag, etags)
+	})
+}
+
+func (h *tieredHealer) run(
+	ctx context.Context,
+	key Key,
+	source, dst Cache,
+	wantETag string,
+	etags *metadatadb.Map[Key, string],
+) {
+	logger := logging.FromContext(ctx)
+	ctx, cancel := context.WithTimeout(ctx, tieredHealTimeout)
+	defer cancel()
+
+	// Version-pinned full read: if the source has moved past wantETag, IfMatch
+	// fails and we skip rather than heal a version we didn't intend to.
+	r, headers, err := source.Open(ctx, key, IfMatch(wantETag))
+	if err != nil {
+		logger.WarnContext(ctx, "Tiered: heal source read failed", "key", key, "etag", wantETag, "error", err)
+		return
+	}
+	defer func() {
+		if err := r.Close(); err != nil {
+			logger.WarnContext(ctx, "Tiered: heal source close failed", "key", key, "error", err)
+		}
+	}()
+	if headers.Get(ETagKey) != wantETag {
+		return
+	}
+
+	w, err := dst.Create(ctx, key, headers, 0, backfillCreateOptions(headers)...) // 0 → cache's max TTL
+	if err != nil {
+		logger.WarnContext(ctx, "Tiered: heal writer create failed", "key", key, "error", err)
+		return
+	}
+	if _, err := io.Copy(w, r); err != nil {
+		logger.WarnContext(ctx, "Tiered: heal copy failed", "key", key, "error", errors.Join(err, w.Abort(err)))
+		return
+	}
+	// Re-check the tracked version right before committing to narrow the window
+	// where a concurrent newer write-through could be clobbered. A residual
+	// race is self-correcting: the next read finds tier 0 stale against the
+	// tracked etag and invalidates it.
+	if got, ok := etags.Get(key); !ok || got != wantETag {
+		if err := w.Abort(errHealSuperseded); err != nil {
+			logger.WarnContext(ctx, "Tiered: heal abort failed", "key", key, "error", err)
+		}
+		return
+	}
+	if err := w.Close(); err != nil {
+		logger.WarnContext(ctx, "Tiered: heal commit failed", "key", key, "error", err)
+		return
+	}
+	logger.DebugContext(ctx, "Tiered: healed divergent tier 0", "key", key, "etag", wantETag)
+}
+
 // backfillReadCloser tees reads from src into dst asynchronously. Chunks are
 // sent to a background goroutine via a buffered channel so the Read path is
 // never blocked by disk I/O (up to ~32 MB of buffer). If the full stream is
@@ -543,7 +710,7 @@ func (t Tiered) Namespace(namespace Namespace) Cache {
 	for i, c := range t.caches {
 		namespaced[i] = c.Namespace(namespace)
 	}
-	return Tiered{caches: namespaced, metadata: t.metadata, etags: tieredETags(t.metadata, namespace), namespace: namespace}
+	return Tiered{caches: namespaced, metadata: t.metadata, etags: tieredETags(t.metadata, namespace), namespace: namespace, healer: t.healer}
 }
 
 // ListNamespaces returns unique namespaces from all underlying caches.

internal/cache/tiered_test.go

@@ -634,6 +634,104 @@ func TestTieredDivergentValidatorPermutations(t *testing.T) {
 	}
 }
 
+// eventually polls fn until it returns true or the deadline elapses.
+func eventually(t *testing.T, fn func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if fn() {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("condition not met before deadline")
+}
+
+func tierHolds(ctx context.Context, t *testing.T, c cache.Cache, key cache.Key, want []byte) bool {
+	t.Helper()
+	r, _, err := c.Open(ctx, key)
+	if err != nil {
+		return false
+	}
+	got := readAllAndClose(t, r)
+	return string(got) == string(want)
+}
+
+// A ranged pinned read served from a deeper tier heals a stale tier 0
+// asynchronously, since a partial body can't be backfilled (#368).
+func TestTieredRangedReadHealsDivergentTier0(t *testing.T) {
+	pinned := []byte("abcdefghij")
+	stale := []byte("0123456789")
+
+	for _, validator := range []struct {
+		name string
+		opts []cache.Option
+	}{
+		{"IfRange", []cache.Option{cache.Range(2, 6), cache.IfRange(`"pinned-etag"`)}},
+		{"IfMatch", []cache.Option{cache.Range(2, 6), cache.IfMatch(`"pinned-etag"`)}},
+	} {
+		t.Run(validator.name, func(t *testing.T) {
+			_, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug})
+			store := newMetadataStore(ctx)
+			lower, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour})
+			assert.NoError(t, err)
+			upper, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour})
+			assert.NoError(t, err)
+			tiered := cache.MaybeNewTiered(ctx, []cache.Cache{lower, upper}, store)
+			defer tiered.Close()
+
+			key := cache.NewKey("heal-ranged")
+			// Replace an existing version through tiered so the etag tracker
+			// records the new authoritative version (the tracker only records
+			// replacements), then diverge tier 0 to a stale version.
+			seedTier(ctx, t, tiered, key, stale, "stale-etag")
+			seedTier(ctx, t, tiered, key, pinned, "pinned-etag")
+			seedTier(ctx, t, lower, key, stale, "stale-etag")
+
+			r, headers, err := tiered.Open(ctx, key, validator.opts...)
+			assert.NoError(t, err)
+			assert.Equal(t, []byte("cdef"), readAllAndClose(t, r))
+			assert.Equal(t, `"pinned-etag"`, headers.Get(cache.ETagKey))
+
+			// The async refresh converges tier 0 to the pinned version.
+			eventually(t, func() bool { return tierHolds(ctx, t, lower, key, pinned) })
+		})
+	}
+}
+
+// A ranged pinned read must not overwrite a tier 0 that is legitimately newer
+// than a lagging deeper tier (#368).
+func TestTieredRangedReadKeepsNewerTier0(t *testing.T) {
+	newer := []byte("abcdefghij")
+	lagging := []byte("0123456789")
+
+	_, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug})
+	store := newMetadataStore(ctx)
+	lower, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour})
+	assert.NoError(t, err)
+	upper, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour})
+	assert.NoError(t, err)
+	tiered := cache.MaybeNewTiered(ctx, []cache.Cache{lower, upper}, store)
+	defer tiered.Close()
+
+	key := cache.NewKey("heal-newer-tier0")
+	// Tier 0 holds the authoritative (newer) version; the deeper tier lags.
+	seedTier(ctx, t, lower, key, newer, "newer-etag")
+	seedTier(ctx, t, upper, key, lagging, "lagging-etag")
+	assert.NoError(t, tieredETags(store, "").Set(key, `"newer-etag"`))
+
+	// A range pinned to the lagging version is served from the deeper tier, but
+	// the tracked etag names tier 0's newer version, so no heal fires.
+	r, headers, err := tiered.Open(ctx, key, cache.Range(2, 6), cache.IfRange(`"lagging-etag"`))
+	assert.NoError(t, err)
+	assert.Equal(t, []byte("2345"), readAllAndClose(t, r))
+	assert.Equal(t, `"lagging-etag"`, headers.Get(cache.ETagKey))
+
+	// Give any (erroneous) heal a chance to run, then confirm tier 0 is intact.
+	time.Sleep(100 * time.Millisecond)
+	assert.True(t, tierHolds(ctx, t, lower, key, newer))
+}
+
 func TestTieredDivergentValidatorProbeErrors(t *testing.T) {
 	_, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug})
 	lower, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour})