perf(cache): fan out large ranged reads through the remote tier

internal/cache/object_window.go

@@ -0,0 +1,62 @@
+package cache
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/alecthomas/errors"
+
+	"github.com/block/cachew/client"
+)
+
+// objectWindow adapts a byte window of a pinned object revision to
+// [client.RangeReader], letting ParallelGet drive parallel range downloads.
+// Range offsets are relative to the window; openRange receives absolute
+// object offsets and must pin every request to the window's revision, so
+// discovery and sub-range requests can never splice revisions. Response
+// headers are synthesized from the pinned values: the backend enforces the
+// real preconditions, surfacing violations as read errors. The window's
+// identity is bound at construction, so the Key argument is ignored.
+type objectWindow struct {
+	openRange func(ctx context.Context, start, length int64) (io.ReadCloser, error)
+	start     int64 // window offset within the object
+	length    int64 // window size in bytes
+	etag      string
+}
+
+// newCacheObjectWindow returns a window over [start, start+length) of the
+// pinned revision of key in c, served by ETag-conditional ranged Opens.
+func newCacheObjectWindow(c client.RangeReader, key Key, start, length int64, etag string) *objectWindow {
+	return &objectWindow{
+		openRange: func(ctx context.Context, start, length int64) (io.ReadCloser, error) {
+			rc, _, err := c.Open(ctx, key, Range(start, start+length), IfMatch(etag))
+			return rc, errors.WithStack(err)
+		},
+		start:  start,
+		length: length,
+		etag:   etag,
+	}
+}
+
+func (w *objectWindow) Open(ctx context.Context, _ Key, opts ...Option) (io.ReadCloser, http.Header, error) {
+	start, length, outcome := NewRequestOptions(opts...).ResolveRange(w.length, w.etag)
+	headers := http.Header{}
+	if w.etag != "" {
+		headers.Set(ETagKey, w.etag)
+	}
+	switch outcome {
+	case RangeNotSatisfiable:
+		headers.Set("Content-Range", fmt.Sprintf("bytes */%d", w.length))
+		return nil, headers, errors.WithStack(ErrRangeNotSatisfiable)
+	case RangePartial:
+		headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, w.length))
+	case RangeFull:
+	}
+	rc, err := w.openRange(ctx, w.start+start, length)
+	if err != nil {
+		return nil, nil, err
+	}
+	return rc, headers, nil
+}

internal/cache/remote.go

@@ -3,7 +3,9 @@ package cache
 import (
 	"context"
 	"io"
+	"math"
 	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/alecthomas/errors"
@@ -38,11 +40,72 @@ func (r *Remote) Namespace(namespace Namespace) Cache {
 	return &Remote{c: r.c.Namespace(namespace)}
 }
 
+// Remote ranged reads at least 2*remoteRangeChunkSize long fan out into
+// parallel sub-range requests, because a single HTTP stream is limited to a
+// fraction of the available bandwidth. Chunks stay below the remote's own S3
+// large-range threshold (2*minRangePartSize) so each sub-range maps to a
+// single upstream request rather than fanning out twice. Memory is bounded by
+// the reader's reorder window: 2*concurrency*chunkSize per read.
+const (
+	remoteRangeChunkSize   = minRangePartSize
+	remoteRangeConcurrency = 8
+)
+
 func (r *Remote) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) {
+	if remoteRangeMayFanOut(NewRequestOptions(opts...)) {
+		return r.parallelRangedOpen(ctx, key, opts)
+	}
 	rc, h, err := r.c.Open(ctx, key, opts...)
 	return rc, h, errors.WithStack(err)
 }
 
+// remoteRangeMayFanOut reports whether the request's Range could span enough
+// bytes to be worth a Stat plus parallel fan-out. The raw range spec is
+// resolved against an unbounded object, so explicit and suffix ranges report
+// their exact requested length and open-ended ranges an effectively infinite
+// one; If-Range gating needs the stored ETag and is deferred to the resolve
+// against the real object.
+func remoteRangeMayFanOut(ro RequestOptions) bool {
+	if ro.Range == "" {
+		return false
+	}
+	ro.IfRange = ""
+	_, length, outcome := ro.ResolveRange(math.MaxInt64, "")
+	return outcome == RangePartial && length >= 2*remoteRangeChunkSize
+}
+
+// parallelRangedOpen serves a large ranged read with parallel sub-range
+// requests pinned to the stored ETag. A preliminary Stat resolves the
+// request's conditionals and range against the object's real size and
+// revision; requests the policy cannot pin or split degrade to a single
+// delegated stream.
+func (r *Remote) parallelRangedOpen(ctx context.Context, key Key, opts []Option) (io.ReadCloser, http.Header, error) {
+	headers, err := r.c.Stat(ctx, key, opts...)
+	if err != nil {
+		return nil, headers, errors.WithStack(err)
+	}
+	size, sizeErr := strconv.ParseInt(headers.Get("Content-Length"), 10, 64)
+	etag := headers.Get(ETagKey)
+	if sizeErr != nil || etag == "" {
+		rc, h, err := r.c.Open(ctx, key, opts...)
+		return rc, h, errors.WithStack(err)
+	}
+	start, length, partial, rangeErr := rangeShortCircuit(headers, size, opts)
+	if rangeErr != nil {
+		return nil, headers, errors.WithStack(rangeErr)
+	}
+	if !partial || length < 2*remoteRangeChunkSize {
+		rc, h, err := r.c.Open(ctx, key, opts...)
+		return rc, h, errors.WithStack(err)
+	}
+	window := newCacheObjectWindow(r.c, key, start, length, etag)
+	rc, err := client.ParallelGetReader(ctx, window, Key{}, remoteRangeChunkSize, remoteRangeConcurrency)
+	if err != nil {
+		return nil, nil, errors.WithStack(err)
+	}
+	return rc, headers, nil
+}
+
 func (r *Remote) Stat(ctx context.Context, key Key, opts ...Option) (http.Header, error) {
 	return errors.WithStack2(r.c.Stat(ctx, key, opts...))
 }

internal/cache/remote_test.go

@@ -1,11 +1,14 @@
 package cache_test
 
 import (
+	"bytes"
 	"io"
 	"log/slog"
+	"math/rand"
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -82,6 +85,125 @@ func TestRemoteInvalidateSkipsRemoteAuthoritativeTier(t *testing.T) {
 	assert.Equal(t, content, data)
 }
 
+// countingHandler wraps an http.Handler, counting HEAD requests and ranged
+// GETs, and how many ranged GETs lack an If-Match revision pin.
+type countingHandler struct {
+	http.Handler
+	heads        atomic.Int32
+	rangedGets   atomic.Int32
+	unpinnedGets atomic.Int32
+}
+
+func (h *countingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch {
+	case r.Method == http.MethodHead:
+		h.heads.Add(1)
+	case r.Method == http.MethodGet && r.Header.Get("Range") != "":
+		h.rangedGets.Add(1)
+		if r.Header.Get("If-Match") == "" {
+			h.unpinnedGets.Add(1)
+		}
+	}
+	h.Handler.ServeHTTP(w, r)
+}
+
+func newCountingRemote(t *testing.T) (cache.Cache, *countingHandler) {
+	t.Helper()
+	ctx := t.Context()
+	_, ctx = logging.Configure(ctx, logging.Config{Level: slog.LevelError})
+	memCache, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 64, MaxTTL: time.Hour})
+	assert.NoError(t, err)
+	t.Cleanup(func() { memCache.Close() })
+
+	mux := http.NewServeMux()
+	_, err = strategy.NewAPIV1(ctx, struct{}{}, memCache, mux)
+	assert.NoError(t, err)
+	counting := &countingHandler{Handler: mux}
+	ts := httptest.NewServer(counting)
+	t.Cleanup(ts.Close)
+
+	remote := cache.NewRemote(ts.URL, nil).Namespace("test")
+	t.Cleanup(func() { remote.Close() })
+	return remote, counting
+}
+
+func TestRemoteLargeRangedReadFansOut(t *testing.T) {
+	ctx := t.Context()
+	remote, counting := newCountingRemote(t)
+
+	data := make([]byte, 12<<20)
+	_, err := rand.New(rand.NewSource(42)).Read(data)
+	assert.NoError(t, err)
+
+	key := cache.NewKey("remote-fan-out")
+	w, err := remote.Create(ctx, key, nil, time.Hour)
+	assert.NoError(t, err)
+	_, err = w.Write(data)
+	assert.NoError(t, err)
+	assert.NoError(t, w.Close())
+
+	r, headers, err := remote.Open(ctx, key, cache.Range(1<<20, 11<<20))
+	assert.NoError(t, err)
+	got, err := io.ReadAll(r)
+	assert.NoError(t, err)
+	assert.NoError(t, r.Close())
+
+	assert.True(t, bytes.Equal(data[1<<20:11<<20], got), "reassembled range differs from original")
+	assert.Equal(t, "bytes 1048576-11534335/12582912", headers.Get("Content-Range"))
+	assert.Equal(t, "10485760", headers.Get("Content-Length"))
+	assert.True(t, counting.rangedGets.Load() >= 3,
+		"expected sub-range fan-out, got %d ranged GETs", counting.rangedGets.Load())
+	assert.Equal(t, int32(0), counting.unpinnedGets.Load(), "every sub-range request must carry If-Match")
+}
+
+func TestRemoteSmallRangedReadStaysSingleStream(t *testing.T) {
+	ctx := t.Context()
+	remote, counting := newCountingRemote(t)
+
+	data := bytes.Repeat([]byte("abcdefgh"), 1<<20/8)
+	key := cache.NewKey("remote-small-range")
+	w, err := remote.Create(ctx, key, nil, time.Hour)
+	assert.NoError(t, err)
+	_, err = w.Write(data)
+	assert.NoError(t, err)
+	assert.NoError(t, w.Close())
+
+	r, _, err := remote.Open(ctx, key, cache.Range(16, 4096))
+	assert.NoError(t, err)
+	got, err := io.ReadAll(r)
+	assert.NoError(t, err)
+	assert.NoError(t, r.Close())
+
+	assert.True(t, bytes.Equal(data[16:4096], got), "range body differs from original")
+	assert.Equal(t, int32(0), counting.heads.Load(), "small ranges must not pay a Stat")
+	assert.Equal(t, int32(1), counting.rangedGets.Load(), "small ranges must use a single request")
+}
+
+func TestRemoteLargeRangedReadIfRangeMissServesFullBody(t *testing.T) {
+	ctx := t.Context()
+	remote, counting := newCountingRemote(t)
+
+	data := make([]byte, 9<<20)
+	_, err := rand.New(rand.NewSource(7)).Read(data)
+	assert.NoError(t, err)
+
+	key := cache.NewKey("remote-if-range-miss")
+	w, err := remote.Create(ctx, key, nil, time.Hour)
+	assert.NoError(t, err)
+	_, err = w.Write(data)
+	assert.NoError(t, err)
+	assert.NoError(t, w.Close())
+
+	r, _, err := remote.Open(ctx, key, cache.Range(0, 9<<20), cache.IfRange(`"other-revision"`))
+	assert.NoError(t, err)
+	got, err := io.ReadAll(r)
+	assert.NoError(t, err)
+	assert.NoError(t, r.Close())
+
+	assert.True(t, bytes.Equal(data, got), "If-Range miss must serve the full representation")
+	assert.Equal(t, int32(1), counting.rangedGets.Load(), "an unpinnable range must degrade to a single stream")
+}
+
 func TestRemoteCacheSoak(t *testing.T) {
 	if os.Getenv("SOAK_TEST") == "" {
 		t.Skip("Skipping soak test; set SOAK_TEST=1 to run")

internal/cache/s3_parallel_get.go

@@ -2,9 +2,7 @@ package cache
 
 import (
 	"context"
-	"fmt"
 	"io"
-	"net/http"
 
 	"github.com/alecthomas/errors"
 	"github.com/minio/minio-go/v7"
@@ -23,7 +21,7 @@ const minRangePartSize int64 = 4 << 20
 func (s *S3) parallelGetReader(ctx context.Context, bucket, objectName string, size int64, etag string) (io.ReadCloser, error) {
 	chunkSize := int64(s.config.DownloadPartSizeMB) << 20                                      // #nosec G115 -- DownloadPartSizeMB is a small operator-supplied tuning value.
 	if concurrency := int(s.config.DownloadConcurrency); size > chunkSize && concurrency > 1 { // #nosec G115 -- DownloadConcurrency is a small operator-supplied tuning value.
-		window := &s3ObjectWindow{s3: s, bucket: bucket, objectName: objectName, start: 0, length: size, etag: etag}
+		window := s.objectWindow(bucket, objectName, 0, size, etag)
 		return client.ParallelGetReader(ctx, window, Key{}, chunkSize, concurrency) //nolint:wrapcheck
 	}
 	obj, err := s.client.GetObject(ctx, bucket, objectName, minio.GetObjectOptions{})
@@ -46,10 +44,24 @@ func (s *S3) rangedGetReader(ctx context.Context, bucket, objectName string, sta
 	}
 	chunkSize := (length + concurrency - 1) / concurrency
 	chunkSize = min(max(chunkSize, minRangePartSize), int64(s.config.DownloadPartSizeMB)<<20) // #nosec G115 -- DownloadPartSizeMB is a small operator-supplied tuning value.
-	window := &s3ObjectWindow{s3: s, bucket: bucket, objectName: objectName, start: start, length: length, etag: etag}
+	window := s.objectWindow(bucket, objectName, start, length, etag)
 	return client.ParallelGetReader(ctx, window, Key{}, chunkSize, int(concurrency)) //nolint:wrapcheck
 }
 
+// objectWindow returns a window over [start, start+length) of the pinned S3
+// object revision, served by direct sub-range GETs so chunk requests bypass
+// S3.Open's per-call stat and range policy.
+func (s *S3) objectWindow(bucket, objectName string, start, length int64, etag string) *objectWindow {
+	return &objectWindow{
+		openRange: func(ctx context.Context, start, length int64) (io.ReadCloser, error) {
+			return s.rangeGetReader(ctx, bucket, objectName, start, length, etag)
+		},
+		start:  start,
+		length: length,
+		etag:   etag,
+	}
+}
+
 // rangeGetReader returns an io.ReadCloser for a single byte range of an S3
 // object, pinned to etag (when non-empty) so the read sees a consistent
 // object revision.
@@ -69,42 +81,3 @@ func (s *S3) rangeGetReader(ctx context.Context, bucket, objectName string, star
 	}
 	return &s3Reader{obj: obj}, nil
 }
-
-// s3ObjectWindow adapts a byte window of a pinned S3 object revision to
-// [client.RangeReader], letting ParallelGet drive parallel S3 downloads.
-// Range offsets are relative to the window, and every request carries an
-// If-Match for the pinned etag regardless of the supplied options, so
-// discovery and sub-range requests can never splice revisions. Response
-// headers are synthesized from the pinned values: minio enforces the real
-// preconditions (SetRange, SetMatchETag) at the protocol level, surfacing
-// violations as read errors. The window's identity is bound in the struct, so
-// the Key argument is ignored.
-type s3ObjectWindow struct {
-	s3         *S3
-	bucket     string
-	objectName string
-	start      int64 // window offset within the object
-	length     int64 // window size in bytes
-	etag       string
-}
-
-func (w *s3ObjectWindow) Open(ctx context.Context, _ Key, opts ...Option) (io.ReadCloser, http.Header, error) {
-	start, length, outcome := NewRequestOptions(opts...).ResolveRange(w.length, w.etag)
-	headers := http.Header{}
-	if w.etag != "" {
-		headers.Set(ETagKey, w.etag)
-	}
-	switch outcome {
-	case RangeNotSatisfiable:
-		headers.Set("Content-Range", fmt.Sprintf("bytes */%d", w.length))
-		return nil, headers, errors.WithStack(ErrRangeNotSatisfiable)
-	case RangePartial:
-		headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, w.length))
-	case RangeFull:
-	}
-	rc, err := w.s3.rangeGetReader(ctx, w.bucket, w.objectName, w.start+start, length, w.etag)
-	if err != nil {
-		return nil, nil, err
-	}
-	return rc, headers, nil
-}