Use HTTP range requests for uncompressed layer fetches in libindex

internal/httputil/rangereaderat.go

@@ -0,0 +1,151 @@
+package httputil
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+// ponytail: read-ahead is a single global buffer per RangeReaderAt; concurrent
+// ReadAt calls serialize on mu. Upgrade path: per-chunk cache keyed by offset.
+const readAheadSize = 256 * 1024
+
+// RangeReaderAt is an io.ReaderAt backed by HTTP range requests.
+type RangeReaderAt struct {
+	client  *http.Client
+	url     string
+	headers http.Header
+	size    int64
+
+	mu     sync.Mutex
+	buf    []byte
+	bufOff int64
+}
+
+// NewRangeReaderAt returns a RangeReaderAt for the blob at url with the given
+// size.
+//
+// headers are copied into each request. size must be the total blob size in
+// bytes.
+func NewRangeReaderAt(client *http.Client, url string, headers http.Header, size int64) *RangeReaderAt {
+	h := headers.Clone()
+	if h == nil {
+		h = make(http.Header)
+	}
+	return &RangeReaderAt{
+		client:  client,
+		url:     url,
+		headers: h,
+		size:    size,
+	}
+}
+
+// Size reports the total blob size.
+func (r *RangeReaderAt) Size() int64 { return r.size }
+
+// ReadAt implements io.ReaderAt.
+func (r *RangeReaderAt) ReadAt(p []byte, off int64) (int, error) {
+	if off < 0 {
+		return 0, fmt.Errorf("httputil: ReadAt: negative offset")
+	}
+	if off >= r.size {
+		return 0, io.EOF
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	start := off
+	wantLen := len(p)
+	total := r.readFromBuffer(p, off)
+	if total == wantLen {
+		return total, nil
+	}
+	if total > 0 {
+		p = p[total:]
+		off += int64(total)
+	}
+
+	fetch := int64(len(p))
+	if fetch < readAheadSize {
+		fetch = readAheadSize
+	}
+	if off+fetch > r.size {
+		fetch = r.size - off
+	}
+
+	data, err := r.fetch(off, fetch)
+	if err != nil {
+		if total > 0 {
+			return total, err
+		}
+		return 0, err
+	}
+	r.bufOff = off
+	r.buf = data
+
+	total += copy(p, r.buf)
+	if total < wantLen && start+int64(total) >= r.size {
+		return total, io.EOF
+	}
+	return total, nil
+}
+
+func (r *RangeReaderAt) readFromBuffer(p []byte, off int64) int {
+	if len(r.buf) == 0 {
+		return 0
+	}
+	end := r.bufOff + int64(len(r.buf))
+	if off < r.bufOff || off >= end {
+		return 0
+	}
+	return copy(p, r.buf[off-r.bufOff:])
+}
+
+func (r *RangeReaderAt) fetch(off, n int64) ([]byte, error) {
+	req, err := http.NewRequest(http.MethodGet, r.url, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header = r.headers.Clone()
+	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", off, off+n-1))
+
+	resp, err := r.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	switch resp.StatusCode {
+	case http.StatusPartialContent:
+		return io.ReadAll(resp.Body)
+	case http.StatusOK:
+		// ponytail: inconsistent server ignored Range; skip to offset in body.
+		if _, err := io.CopyN(io.Discard, resp.Body, off); err != nil {
+			return nil, err
+		}
+		return io.ReadAll(io.LimitReader(resp.Body, n))
+	}
+	return nil, CheckResponse(resp, http.StatusPartialContent, http.StatusOK)
+}
+
+// ParseContentRangeTotal extracts the total size from a Content-Range header.
+//
+// For example, "bytes 0-15/12345" returns 12345.
+func ParseContentRangeTotal(h string) (int64, error) {
+	if !strings.HasPrefix(h, "bytes ") {
+		return 0, fmt.Errorf("httputil: invalid Content-Range: %q", h)
+	}
+	_, totalText, ok := strings.Cut(h, "/")
+	if !ok {
+		return 0, fmt.Errorf("httputil: invalid Content-Range: %q", h)
+	}
+	total, err := strconv.ParseInt(totalText, 10, 64)
+	if err != nil {
+		return 0, fmt.Errorf("httputil: invalid Content-Range total: %w", err)
+	}
+	return total, nil
+}

internal/httputil/rangereaderat_test.go

@@ -0,0 +1,74 @@
+package httputil
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestRangeReaderAt(t *testing.T) {
+	data := bytes.Repeat([]byte("abcdefghijklmnop"), 32*1024) // 512 KiB
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.ServeContent(w, r, "blob", time.Now(), bytes.NewReader(data))
+	}))
+	t.Cleanup(srv.Close)
+
+	ra := NewRangeReaderAt(srv.Client(), srv.URL, nil, int64(len(data)))
+
+	// Read across a chunk boundary to exercise read-ahead.
+	p := make([]byte, 300*1024)
+	n, err := ra.ReadAt(p, 100*1024)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != len(p) {
+		t.Fatalf("got %d bytes, want %d", n, len(p))
+	}
+	if !bytes.Equal(p, data[100*1024:100*1024+len(p)]) {
+		t.Fatal("data mismatch")
+	}
+
+	// EOF at end.
+	_, err = ra.ReadAt(make([]byte, 1), int64(len(data)))
+	if err != io.EOF {
+		t.Fatalf("got %v, want EOF", err)
+	}
+}
+
+func TestRangeReaderAtBufferHit(t *testing.T) {
+	var requests int
+	data := bytes.Repeat([]byte{0x42}, readAheadSize*2)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		http.ServeContent(w, r, "blob", time.Now(), bytes.NewReader(data))
+	}))
+	t.Cleanup(srv.Close)
+
+	ra := NewRangeReaderAt(srv.Client(), srv.URL, nil, int64(len(data)))
+
+	buf := make([]byte, 512)
+	if _, err := ra.ReadAt(buf, 0); err != nil {
+		t.Fatal(err)
+	}
+	before := requests
+	if _, err := ra.ReadAt(buf, 256); err != nil {
+		t.Fatal(err)
+	}
+	if requests != before {
+		t.Fatalf("second read triggered HTTP request: %d -> %d", before, requests)
+	}
+}
+
+func TestParseContentRangeTotal(t *testing.T) {
+	t.Parallel()
+	got, err := ParseContentRangeTotal("bytes 0-15/12345")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != 12345 {
+		t.Fatalf("got %d, want 12345", got)
+	}
+}

internal/zreader/zreader.go

@@ -89,9 +89,9 @@ var (
 // DetectCompression reports the compression type indicated based on the header
 // contained in the passed byte slice.
 //
-// "CmpNone" is returned if all detectors report false, but it's possible that
+// "KindNone" is returned if all detectors report false, but it's possible that
 // it's just a scheme unsupported by this package.
-func detectCompression(b []byte) Compression {
+func DetectCompression(b []byte) Compression {
 	t := make([]byte, len(b))
 	for c, d := range detectors {
 		n, l := copy(t, b), len(d.Mask)
@@ -153,7 +153,7 @@ func detect(r io.Reader) (io.ReadCloser, Compression, error) {
 	//
 	// All the return types are a little different, so they're handled in the
 	// switch arms.
-	switch c := detectCompression(b); c {
+	switch c := DetectCompression(b); c {
 	case KindGzip:
 		z, err := gzip.NewReader(br)
 		return z, c, err

libindex/fetcher.go

@@ -125,6 +125,21 @@ func (a *RemoteFetchArena) fetchInto(ctx context.Context, l *claircore.Layer, cl
 			span.End()
 		}()
 
+		// Try HTTP range fetch for uncompressed layers.
+		if ra, ok, err := a.tryNewRangeReaderAt(ctx, desc); err != nil {
+			return err
+		} else if ok {
+			cacheHit = false
+			if err := l.Init(ctx, desc, ra); err != nil {
+				return err
+			}
+			*cl = closeFunc(func() error {
+				runtime.KeepAlive(ra)
+				return l.Close()
+			})
+			return nil
+		}
+
 		// NB This is not closed on purpose. The [io.Closer] populated by this
 		// function holds the pointer until that function is cleaned up. Once
 		// nothing has a copy of this [*os.File], the runtime will run all the
@@ -223,12 +238,8 @@ func (a *RemoteFetchArena) fetchFileForCache(ctx context.Context, desc *claircor
 	var cmpSz writeCounter
 	tr := io.TeeReader(resp.Body, io.MultiWriter(&cmpSz, vh))
 
-	// TODO(hank) All this decompression code could go away, but that would mean
-	// that a buffer file would have to be allocated later, adding additional
-	// disk usage.
-	//
-	// The ultimate solution is to move to a fetcher that proxies to HTTP range
-	// requests.
+	// Compressed layers are fully downloaded and spooled here. Uncompressed
+	// layers use HTTP range requests via fetchInto and [httputil.RangeReaderAt].
 	zr, kind, err := zreader.Detect(tr)
 	if err != nil {
 		return nil, fmt.Errorf("fetcher: error determining compression: %w", err)
@@ -344,6 +355,56 @@ func (a *RemoteFetchArena) Close(_ context.Context) error {
 	return nil
 }
 
+const rangeProbeSize = 16
+
+func (a *RemoteFetchArena) tryNewRangeReaderAt(ctx context.Context, desc *claircore.LayerDescription) (*httputil.RangeReaderAt, bool, error) {
+	if desc.URI == "" {
+		return nil, false, fmt.Errorf("empty uri for layer %v", desc.Digest)
+	}
+	url, err := url.ParseRequestURI(desc.URI)
+	if err != nil {
+		return nil, false, fmt.Errorf("failed to parse remote path uri: %v", err)
+	}
+
+	req := (&http.Request{
+		ProtoMajor: 1,
+		ProtoMinor: 1,
+		Proto:      "HTTP/1.1",
+		Host:       url.Host,
+		Method:     http.MethodGet,
+		URL:        url,
+		Header:     http.Header(desc.Headers).Clone(),
+	}).WithContext(ctx)
+	req.Header.Set("Range", fmt.Sprintf("bytes=0-%d", rangeProbeSize-1))
+
+	resp, err := a.wc.Do(req)
+	if err != nil {
+		return nil, false, nil
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusPartialContent {
+		return nil, false, nil
+	}
+
+	probe, err := io.ReadAll(io.LimitReader(resp.Body, rangeProbeSize))
+	if err != nil {
+		return nil, false, nil
+	}
+	if zreader.DetectCompression(probe) != zreader.KindNone {
+		return nil, false, nil
+	}
+
+	size, err := httputil.ParseContentRangeTotal(resp.Header.Get("Content-Range"))
+	if err != nil {
+		return nil, false, nil
+	}
+
+	// Range mode trusts the registry digest; full verification would require
+	// downloading the entire blob.
+	return httputil.NewRangeReaderAt(a.wc, desc.URI, desc.Headers, size), true, nil
+}
+
 // Realizer returns an indexer.Realizer.
 //
 // The concrete return type is [*FetchProxy].