From d1381a2fc0b1c3904b9e11ac8252f647f8098dc9 Mon Sep 17 00:00:00 2001 From: Elizabeth Worstell Date: Mon, 20 Jul 2026 13:08:45 -0700 Subject: [PATCH] perf(cache): parallelize tiered heal reads from the source tier The tier-0 heal previously pulled the whole object from the source tier as a single stream, so only an S3 source (which fans out internally) saturated the link. The heal now drives client.ParallelGetReader itself, pinned to the healed revision via If-Match, giving any Range-capable source tier parallel throughput and degrading to a single stream for sources without Range/ETag support. Heal chunks stay below the S3 tier's internal large-range fan-out threshold so requests are never amplified twice. --- internal/cache/tiered.go | 43 +++++++++++++++++++-- internal/cache/tiered_test.go | 70 +++++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 4 deletions(-) diff --git a/internal/cache/tiered.go b/internal/cache/tiered.go index 3a0bfa4..b65250c 100644 --- a/internal/cache/tiered.go +++ b/internal/cache/tiered.go @@ -12,6 +12,7 @@ import ( "github.com/alecthomas/errors" + "github.com/block/cachew/client" "github.com/block/cachew/internal/logging" "github.com/block/cachew/internal/metadatadb" ) @@ -450,20 +451,54 @@ func (t Tiered) healTier0(reqCtx context.Context, key Key, source Cache, servedE }) } +// Heal chunks stay below the S3 tier's internal large-range fan-out threshold +// (2*minRangePartSize) so each chunk maps to a single upstream request rather +// than fanning out twice. +const ( + tieredHealChunkSize = minRangePartSize + tieredHealConcurrency = 8 +) + +// healSource pins every heal read to the revision being backfilled, so +// discovery and chunk requests fail with ErrPreconditionFailed instead of +// splicing in a concurrent rewrite. +type healSource struct { + c Cache + etag string +} + +func (s healSource) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) { + r, h, err := s.c.Open(ctx, key, append(opts, IfMatch(s.etag))...) + return r, h, errors.WithStack(err) +} + +// backfillTier0FromSource re-fetches the whole object from source and writes +// it to tier 0. Heals pull from a deeper (often network) tier where a single +// stream is bandwidth-limited, so the read fans out into parallel chunk +// requests; discovery degrades to a single stream when the source lacks Range +// or ETag support. func (t Tiered) backfillTier0FromSource(ctx context.Context, key Key, source Cache, wantETag string) { logger := logging.FromContext(ctx) ctx, cancel := context.WithTimeout(ctx, tieredHealTimeout) defer cancel() - r, headers, err := source.Open(ctx, key, IfMatch(wantETag)) + // If-Match lets a tiered source probe past stale front tiers for the wanted + // revision; a precondition failure means no tier holds it anymore. + headers, err := source.Stat(ctx, key, IfMatch(wantETag)) + if errors.Is(err, ErrPreconditionFailed) { + return + } + if err != nil { + logger.WarnContext(ctx, "Tiered: ranged heal source stat failed", "key", key, "etag", wantETag, "error", err) + return + } + + r, err := client.ParallelGetReader(ctx, healSource{c: source, etag: wantETag}, key, tieredHealChunkSize, tieredHealConcurrency) if err != nil { logger.WarnContext(ctx, "Tiered: ranged heal source read failed", "key", key, "etag", wantETag, "error", err) return } defer discardTieredReader(ctx, key, r) - if headers.Get(ETagKey) != wantETag { - return - } w, err := t.caches[0].Create(ctx, key, headers, 0, backfillCreateOptions(headers)...) // 0 → cache's max TTL if err != nil { diff --git a/internal/cache/tiered_test.go b/internal/cache/tiered_test.go index 26cc651..695c5e6 100644 --- a/internal/cache/tiered_test.go +++ b/internal/cache/tiered_test.go @@ -1,6 +1,7 @@ package cache_test import ( + "bytes" "context" "fmt" "io" @@ -700,6 +701,75 @@ func TestTieredRangedReadHealsDivergentTier0(t *testing.T) { } } +type countingCache struct { + cache.Cache + rangedOpens atomic.Int32 +} + +func (c *countingCache) Open(ctx context.Context, key cache.Key, opts ...cache.Option) (io.ReadCloser, http.Header, error) { + if cache.NewRequestOptions(opts...).Range != "" { + c.rangedOpens.Add(1) + } + return c.Cache.Open(ctx, key, opts...) //nolint:wrapcheck +} + +func TestTieredHealFansOutLargeObjects(t *testing.T) { + pinned := bytes.Repeat([]byte("abcdefgh"), 9<<20/8) + stale := []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) + upperMem, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour}) + assert.NoError(t, err) + upper := &countingCache{Cache: upperMem} + tiered := cache.MaybeNewTiered(ctx, []cache.Cache{lower, upper}, store) + defer tiered.Close() + + key := cache.NewKey("heal-fan-out") + 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, _, err := tiered.Open(ctx, key, cache.Range(2, 6), cache.IfRange(`"pinned-etag"`)) + assert.NoError(t, err) + assert.Equal(t, pinned[2:6], readAllAndClose(t, r)) + + eventually(t, func() bool { return tierHolds(ctx, t, lower, key, pinned, `"pinned-etag"`) }) + assert.True(t, upper.rangedOpens.Load() >= 3, "expected fan-out, got %d ranged opens", upper.rangedOpens.Load()) +} + +func TestTieredHealProbesPastStaleSourceTier(t *testing.T) { + pinned := []byte("abcdefghij") + stale := []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) + sourceFront, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour}) + assert.NoError(t, err) + sourceBack, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1024, MaxTTL: time.Hour}) + assert.NoError(t, err) + source := cache.MaybeNewTiered(ctx, []cache.Cache{sourceFront, sourceBack}, newMetadataStore(ctx)) + defer source.Close() + tiered := cache.MaybeNewTiered(ctx, []cache.Cache{lower, source}, store) + defer tiered.Close() + + key := cache.NewKey("heal-stale-source-tier") + seedTier(ctx, t, sourceFront, key, stale, "stale-etag") + seedTier(ctx, t, sourceBack, key, pinned, "pinned-etag") + seedTier(ctx, t, lower, key, stale, "stale-etag") + assert.NoError(t, tieredETags(store, "").Set(key, `"pinned-etag"`)) + + r, _, err := tiered.Open(ctx, key, cache.Range(2, 6), cache.IfRange(`"pinned-etag"`)) + assert.NoError(t, err) + assert.Equal(t, pinned[2:6], readAllAndClose(t, r)) + + eventually(t, func() bool { return tierHolds(ctx, t, lower, key, pinned, `"pinned-etag"`) }) +} + func TestTieredRangedReadKeepsNewerTier0(t *testing.T) { newer := []byte("abcdefghij") lagging := []byte("0123456789")