feat(client): stream parallel downloads to pipes

.golangci.yml

@@ -480,7 +480,7 @@ linters:
       # components intentionally forward URLs and construct paths from config.
       - text: "G703:"
         linters: [gosec]
-        path: "(internal/cache/disk\\.go|internal/strategy/git/spool\\.go)"
+        path: "(internal/cache/disk\\.go|internal/strategy/git/spool\\.go|client/spill\\.go)"
       - text: "G704:"
         linters: [gosec]
       - text: "avoid package names that conflict with Go standard library"

client/git.go

@@ -201,12 +201,36 @@ type GitSnapshotMetadata struct {
 // returns the snapshot's freshen metadata, read from the discovery response.
 // Returns os.ErrNotExist when the server has no snapshot available.
 func (c *Client) DownloadGitSnapshot(ctx context.Context, repoURL string, dst io.WriterAt, chunkSize int64, concurrency int) (GitSnapshotMetadata, error) {
+	return c.downloadGitSnapshot(repoURL, func(reader *gitArtifactRangeReader) error {
+		return errors.WithStack(ParallelGet(ctx, reader, NewKey(repoURL), dst, chunkSize, concurrency))
+	})
+}
+
+// DownloadGitSnapshotStream fetches the working-tree snapshot for repoURL and
+// writes it to w as a sequential byte stream, fetching up to concurrency
+// chunkSize-byte ranges concurrently and reordering them through a spill file
+// in spillDir (see [ParallelGetStream]). When concurrency is 1 the single
+// response is streamed directly to w with no spill file; a server without
+// range support falls back to a single full download, which still round-trips
+// through the spill file. It returns the snapshot's freshen metadata, read
+// from the discovery response. Returns os.ErrNotExist when the server has no
+// snapshot available.
+func (c *Client) DownloadGitSnapshotStream(ctx context.Context, repoURL string, w io.Writer, chunkSize int64, concurrency int, spillDir string) (GitSnapshotMetadata, error) {
+	return c.downloadGitSnapshot(repoURL, func(reader *gitArtifactRangeReader) error {
+		return errors.WithStack(ParallelGetStream(ctx, reader, NewKey(repoURL), w, chunkSize, concurrency, spillDir))
+	})
+}
+
+// downloadGitSnapshot resolves repoURL's snapshot endpoint, runs fetch against
+// a metadata-recording RangeReader for it, and returns the recorded freshen
+// metadata.
+func (c *Client) downloadGitSnapshot(repoURL string, fetch func(*gitArtifactRangeReader) error) (GitSnapshotMetadata, error) {
 	endpoint, err := gitEndpointURL(c.baseURL, repoURL, "snapshot.tar.zst")
 	if err != nil {
 		return GitSnapshotMetadata{}, err
 	}
 	reader := &gitArtifactRangeReader{client: c, endpoint: endpoint}
-	if err := ParallelGet(ctx, reader, NewKey(repoURL), dst, chunkSize, concurrency); err != nil {
+	if err := fetch(reader); err != nil {
 		return GitSnapshotMetadata{}, errors.Wrap(err, "download snapshot")
 	}
 	return reader.metadata(), nil

client/git_test.go

@@ -175,26 +175,32 @@ func TestOpenGitBundleNotFound(t *testing.T) {
 	assert.True(t, errors.Is(err, os.ErrNotExist))
 }
 
-func TestDownloadGitSnapshotParallel(t *testing.T) {
-	body := make([]byte, 1000)
-	for i := range body {
-		body[i] = byte(i % 251)
-	}
-	const etag = `"snap-v1"`
-
-	var requests atomic.Int64
+// newSnapshotRangeServer serves body as a ranged snapshot endpoint, counting
+// requests. ServeContent honours Range/If-Range against the ETag, so it
+// returns 206 + Content-Range for the chunked requests ParallelGet makes.
+func newSnapshotRangeServer(t *testing.T, body []byte, etag string, requests *atomic.Int64) *httptest.Server {
+	t.Helper()
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		assert.Equal(t, "/git/github.com/org/repo/snapshot.tar.zst", r.URL.Path)
 		requests.Add(1)
 		w.Header().Set("Content-Type", "application/zstd")
 		w.Header().Set("ETag", etag)
 		w.Header().Set(client.SnapshotCommitHeader, "deadbeef")
 		w.Header().Set(client.BundleURLHeader, "/git/github.com/org/repo/snapshot.bundle?base=deadbeef")
-		// ServeContent honours Range/If-Range against the ETag set above, so it
-		// returns 206 + Content-Range for the chunked requests ParallelGet makes.
 		http.ServeContent(w, r, "snapshot.tar.zst", time.Time{}, bytes.NewReader(body))
 	}))
-	defer srv.Close()
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestDownloadGitSnapshotParallel(t *testing.T) {
+	body := make([]byte, 1000)
+	for i := range body {
+		body[i] = byte(i % 251)
+	}
+
+	var requests atomic.Int64
+	srv := newSnapshotRangeServer(t, body, `"snap-v1"`, &requests)
 
 	api := client.NewWithHTTPClient(srv.URL, srv.Client())
 	var dst bufferAt
@@ -208,6 +214,75 @@ func TestDownloadGitSnapshotParallel(t *testing.T) {
 	assert.True(t, requests.Load() > 1, "expected multiple range requests, got %d", requests.Load())
 }
 
+func TestDownloadGitSnapshotStreamParallel(t *testing.T) {
+	body := make([]byte, 1000)
+	for i := range body {
+		body[i] = byte(i % 251)
+	}
+
+	var requests atomic.Int64
+	srv := newSnapshotRangeServer(t, body, `"snap-v1"`, &requests)
+
+	api := client.NewWithHTTPClient(srv.URL, srv.Client())
+	var out bytes.Buffer
+	meta, err := api.DownloadGitSnapshotStream(context.Background(), "https://github.com/org/repo", &out, 128, 4, t.TempDir())
+	assert.NoError(t, err)
+	assert.Equal(t, body, out.Bytes())
+	assert.Equal(t, "deadbeef", meta.Commit)
+	assert.Equal(t, "/git/github.com/org/repo/snapshot.bundle?base=deadbeef", meta.BundleURL)
+	assert.True(t, requests.Load() > 1, "expected multiple range requests, got %d", requests.Load())
+}
+
+func TestDownloadGitSnapshotStreamSingleRequest(t *testing.T) {
+	body := make([]byte, 1000)
+	for i := range body {
+		body[i] = byte(i % 251)
+	}
+
+	var requests atomic.Int64
+	srv := newSnapshotRangeServer(t, body, `"snap-v1"`, &requests)
+
+	api := client.NewWithHTTPClient(srv.URL, srv.Client())
+	var out bytes.Buffer
+	meta, err := api.DownloadGitSnapshotStream(context.Background(), "https://github.com/org/repo", &out, 128, 1, t.TempDir())
+	assert.NoError(t, err)
+	assert.Equal(t, body, out.Bytes())
+	assert.Equal(t, "deadbeef", meta.Commit)
+	assert.Equal(t, int64(1), requests.Load())
+}
+
+func TestDownloadGitSnapshotStreamFallsBackWithoutRange(t *testing.T) {
+	body := []byte("full body, server ignores ranges")
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		// No ETag and no range handling: always answer the full object with 200,
+		// mimicking an older server. ParallelGetStream must fall back to a
+		// single read through the spill file.
+		w.Header().Set("Content-Type", "application/zstd")
+		w.Header().Set(client.SnapshotCommitHeader, "cafe")
+		_, _ = w.Write(body) //nolint:errcheck
+	}))
+	defer srv.Close()
+
+	api := client.NewWithHTTPClient(srv.URL, srv.Client())
+	var out bytes.Buffer
+	meta, err := api.DownloadGitSnapshotStream(context.Background(), "https://github.com/org/repo", &out, 8, 4, t.TempDir())
+	assert.NoError(t, err)
+	assert.Equal(t, body, out.Bytes())
+	assert.Equal(t, "cafe", meta.Commit)
+}
+
+func TestDownloadGitSnapshotStreamNotFound(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.NotFound(w, r)
+	}))
+	defer srv.Close()
+
+	api := client.NewWithHTTPClient(srv.URL, srv.Client())
+	var out bytes.Buffer
+	_, err := api.DownloadGitSnapshotStream(context.Background(), "https://github.com/org/repo", &out, 8, 4, t.TempDir())
+	assert.True(t, errors.Is(err, os.ErrNotExist))
+}
+
 func TestDownloadGitSnapshotFallsBackWithoutRange(t *testing.T) {
 	body := []byte("full body, server ignores ranges")
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {

client/internal/punchholelayout/layout.go

@@ -0,0 +1,18 @@
+// Package punchholelayout captures the memory layout of the macOS SDK's
+// fpunchhole_t via cgo, so tests can verify that the hand-defined Go mirror
+// in package client stays identical to the C struct.
+package punchholelayout
+
+// Layout describes the size and field layout of a punch-hole argument struct,
+// in bytes.
+type Layout struct {
+	Size           uintptr
+	FlagsOffset    uintptr
+	FlagsSize      uintptr
+	ReservedOffset uintptr
+	ReservedSize   uintptr
+	OffsetOffset   uintptr
+	OffsetSize     uintptr
+	LengthOffset   uintptr
+	LengthSize     uintptr
+}

client/internal/punchholelayout/layout_darwin_cgo.go

@@ -0,0 +1,27 @@
+//go:build darwin && cgo
+
+package punchholelayout
+
+/*
+#include <sys/fcntl.h>
+*/
+import "C"
+
+import "unsafe"
+
+// SDKLayout returns the layout of struct fpunchhole_t as compiled from the
+// SDK's <sys/fcntl.h>.
+func SDKLayout() (Layout, bool) {
+	var v C.fpunchhole_t
+	return Layout{
+		Size:           unsafe.Sizeof(v),
+		FlagsOffset:    unsafe.Offsetof(v.fp_flags),
+		FlagsSize:      unsafe.Sizeof(v.fp_flags),
+		ReservedOffset: unsafe.Offsetof(v.reserved),
+		ReservedSize:   unsafe.Sizeof(v.reserved),
+		OffsetOffset:   unsafe.Offsetof(v.fp_offset),
+		OffsetSize:     unsafe.Sizeof(v.fp_offset),
+		LengthOffset:   unsafe.Offsetof(v.fp_length),
+		LengthSize:     unsafe.Sizeof(v.fp_length),
+	}, true
+}

client/internal/punchholelayout/layout_stub.go

@@ -0,0 +1,9 @@
+//go:build !darwin || !cgo
+
+package punchholelayout
+
+// SDKLayout reports false: without cgo on darwin there is no SDK struct to
+// compare against.
+func SDKLayout() (Layout, bool) {
+	return Layout{}, false
+}

client/parallel_get.go

@@ -91,6 +91,39 @@ func ParallelGet(ctx context.Context, c RangeReader, key Key, dst io.WriterAt, c
 	return errors.Wrap(eg.Wait(), "parallel get")
 }
 
+// ParallelGetStream downloads an object from any Range-capable RangeReader
+// into w as a sequential byte stream, while still fetching chunkSize-byte
+// chunks with up to concurrency concurrent requests. Out-of-order chunks are
+// reordered through an unlinked spill file created in spillDir rather than
+// held in RAM; consumed regions of the spill file are hole-punched where the
+// platform supports it, so its disk footprint tracks the gap between download
+// and consumption. A concurrency of 1 streams the single response directly to
+// w and touches no disk. Revision-safety semantics match [ParallelGet]: a
+// partially written w must be discarded by the caller on failure.
+func ParallelGetStream(ctx context.Context, c RangeReader, key Key, w io.Writer, chunkSize int64, concurrency int, spillDir string) error {
+	// chunkSize is irrelevant to the single-request path, so it is validated
+	// by ParallelGet only when actually chunking.
+	if concurrency <= 1 {
+		rc, _, err := c.Open(ctx, key)
+		if err != nil {
+			return errors.Wrap(err, "parallel get: full read")
+		}
+		_, copyErr := io.Copy(w, rc)
+		return errors.Wrap(errors.Join(copyErr, rc.Close()), "parallel get")
+	}
+
+	sb, err := newSpillBuffer(spillDir)
+	if err != nil {
+		return errors.Wrap(err, "parallel get")
+	}
+	eg, egCtx := errgroup.WithContext(ctx)
+	eg.Go(func() error {
+		return sb.closeWrite(ParallelGet(egCtx, c, key, sb, chunkSize, concurrency))
+	})
+	eg.Go(func() error { return sb.streamTo(egCtx, w) })
+	return errors.Join(eg.Wait(), sb.Close())
+}
+
 func fullRead(ctx context.Context, c RangeReader, key Key, dst io.WriterAt) error {
 	rc, _, err := c.Open(ctx, key)
 	if err != nil {

client/parallel_get_test.go

@@ -14,6 +14,7 @@ import (
 	"testing"
 
 	"github.com/alecthomas/assert/v2"
+	"github.com/alecthomas/errors"
 
 	"github.com/block/cachew/client"
 )
@@ -249,6 +250,62 @@ func TestParallelGetNoETagSizeChangedBetweenRequests(t *testing.T) {
 	assert.Equal(t, c.rewritten, dst.buf)
 }
 
+func TestParallelGetStream(t *testing.T) {
+	data := make([]byte, 1000)
+	for i := range data {
+		data[i] = byte(i % 251)
+	}
+
+	tests := []struct {
+		name        string
+		reader      client.RangeReader
+		concurrency int
+		want        []byte
+		wantErrIs   error
+		wantErrText string
+	}{
+		{name: "MultiChunk", reader: &recordingReader{data: data, etag: `"v1"`}, concurrency: 4, want: data},
+		{name: "SingleWorker", reader: &recordingReader{data: data, etag: `"v1"`}, concurrency: 1, want: data},
+		{name: "NoETagFallback", reader: &noETagReader{data: data}, concurrency: 4, want: data},
+		{name: "Empty", reader: &recordingReader{etag: `"v1"`}, concurrency: 4},
+		{name: "ETagMismatch", reader: &rangeFlipReader{data: data, firstETag: `"v1"`, restETag: `"v2"`},
+			concurrency: 4, wantErrText: "object changed during read"},
+		{name: "PreconditionFailed", reader: &ifMatchFlipReader{data: data, firstETag: `"v1"`, restETag: `"v2"`},
+			concurrency: 4, wantErrIs: client.ErrPreconditionFailed},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var out bytes.Buffer
+			err := client.ParallelGetStream(context.Background(), tt.reader, client.NewKey("k"), &out, 100, tt.concurrency, t.TempDir())
+			switch {
+			case tt.wantErrIs != nil:
+				assert.IsError(t, err, tt.wantErrIs)
+			case tt.wantErrText != "":
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErrText)
+			default:
+				assert.NoError(t, err)
+				assert.Equal(t, tt.want, out.Bytes())
+			}
+		})
+	}
+}
+
+type failingWriter struct{ err error }
+
+func (f *failingWriter) Write([]byte) (int, error) { return 0, f.err }
+
+func TestParallelGetStreamWriterError(t *testing.T) {
+	data := make([]byte, 1000)
+	for i := range data {
+		data[i] = byte(i % 251)
+	}
+	c := &recordingReader{data: data, etag: `"v1"`}
+	sinkErr := errors.New("sink failed")
+	err := client.ParallelGetStream(context.Background(), c, client.NewKey("k"), &failingWriter{err: sinkErr}, 100, 4, t.TempDir())
+	assert.IsError(t, err, sinkErr)
+}
+
 func TestParallelGetClient(t *testing.T) {
 	content := make([]byte, 1000)
 	for i := range content {

client/punchhole_darwin.go

@@ -0,0 +1,34 @@
+//go:build darwin
+
+package client
+
+import (
+	"os"
+	"runtime"
+	"unsafe"
+
+	"github.com/alecthomas/errors"
+	"golang.org/x/sys/unix"
+)
+
+// fpunchhole mirrors struct fpunchhole_t from <sys/fcntl.h>. x/sys/unix
+// defines F_PUNCHHOLE for darwin but neither the argument struct nor a
+// wrapper, so we pass the pointer through FcntlInt the same way
+// unix.FcntlFstore does for F_PREALLOCATE.
+type fpunchhole struct {
+	flags    uint32 // fp_flags: no flags are currently defined
+	reserved uint32
+	offset   int64
+	length   int64
+}
+
+// punchHole deallocates the byte range [off, off+length) of f, leaving a hole
+// that reads as zeros without changing the file size. off and length must be
+// multiples of the filesystem block size. Requires APFS.
+func punchHole(f *os.File, off, length int64) error {
+	arg := fpunchhole{offset: off, length: length}
+	//nolint:gosec // pointers fit in int on darwin; the same pattern unix.FcntlFstore uses
+	_, err := unix.FcntlInt(f.Fd(), unix.F_PUNCHHOLE, int(uintptr(unsafe.Pointer(&arg))))
+	runtime.KeepAlive(&arg)
+	return errors.Wrap(err, "punch hole")
+}