diff --git a/client/parallel_get.go b/client/parallel_get.go new file mode 100644 index 00000000..dad25c27 --- /dev/null +++ b/client/parallel_get.go @@ -0,0 +1,149 @@ +package client + +import ( + "context" + "io" + "net/http" + "strconv" + "strings" + + "github.com/alecthomas/errors" + "golang.org/x/sync/errgroup" +) + +// RangeReader is the subset of cache operations ParallelGet needs: a +// conditional, Range-capable Open. Both *Client and the cache.Cache interface +// satisfy it, so ParallelGet can drive either a remote cache server or a local +// backend. +type RangeReader interface { + Open(ctx context.Context, key Key, opts ...RequestOption) (io.ReadCloser, http.Header, error) +} + +// ParallelGet downloads an object from any Range-capable RangeReader into dst, +// fetching it in chunkSize-byte chunks concurrently (up to concurrency requests +// in flight) and writing each chunk at its offset via dst.WriteAt. Latency-bound +// backends such as a remote cache can saturate bandwidth with overlapping reads. +// +// The first chunk is fetched with a ranged Open, whose response yields both the +// total size (from Content-Range) and the object's ETag; every remaining chunk +// is then requested with IfRange pinned to that ETag. If the object changes +// mid-download, a chunk's ETag will differ and ParallelGet returns an error +// rather than splicing bytes from two revisions. A missing or truncated chunk +// is likewise reported as an error, so a partially written dst must be discarded +// by the caller on failure. An object with no ETag to pin to (e.g. one stored +// before ETags were recorded) cannot be kept revision-safe across chunks, so it +// falls back to a single full read instead of parallelising. +// +// dst is written via concurrent WriteAt calls at non-overlapping offsets; the +// caller owns dst's lifecycle (open, close, cleanup) and need not pre-size it, +// as WriteAt extends the destination. +func ParallelGet(ctx context.Context, c RangeReader, key Key, dst io.WriterAt, chunkSize int64, concurrency int) error { + if chunkSize <= 0 { + return errors.Errorf("parallel get: chunk size must be positive, got %d", chunkSize) + } + concurrency = max(concurrency, 1) + + // Discovery: the first ranged Open delivers chunk zero and reveals the total + // size and ETag used to pin the rest. + rc, headers, err := c.Open(ctx, key, Range(0, chunkSize)) + if errors.Is(err, ErrRangeNotSatisfiable) { + return nil // Empty object: nothing to write. + } + if err != nil { + return errors.Wrap(err, "parallel get: open first chunk") + } + + etag := headers.Get(ETagKey) + total, hasRange := parseContentRangeTotal(headers.Get("Content-Range")) + + // A backend that ignored the range (no Content-Range), or an object that + // fits within the first chunk, is delivered entirely by this response: copy + // it and return, as there is nothing to parallelise. A negative want skips + // the length check when the total size is unknown. + firstLen := min(chunkSize, total) + if !hasRange { + firstLen = -1 + } + if !hasRange || total <= chunkSize { + return errors.Wrap(writeChunkAt(dst, 0, firstLen, rc), "parallel get") + } + + // Subsequent chunks are pinned to the discovery ETag via IfRange. Without a + // validator there is nothing to pin to (IfRange("") is a no-op and an empty + // ETag matches an empty ETag), so chunks could be spliced across a rewrite + // undetected. Objects stored before ETags were recorded fall here, so fall + // back to a single, revision-consistent read rather than parallelising. + if etag == "" { + if err := rc.Close(); err != nil { + return errors.Wrap(err, "parallel get: close discovery reader") + } + full, _, err := c.Open(ctx, key) + if err != nil { + return errors.Wrap(err, "parallel get: full read") + } + return errors.Wrap(writeChunkAt(dst, 0, total, full), "parallel get") + } + + // Multiple chunks: copy the already-open first chunk concurrently with the + // rest rather than blocking on it here. The first goroutine is scheduled + // before the limit can be reached, so it never stalls holding an open body. + numChunks := int((total + chunkSize - 1) / chunkSize) + eg, egCtx := errgroup.WithContext(ctx) + eg.SetLimit(concurrency) + eg.Go(func() error { return writeChunkAt(dst, 0, firstLen, rc) }) + for seq := 1; seq < numChunks; seq++ { + // Stop scheduling once a chunk has failed and cancelled the group. + if egCtx.Err() != nil { + break + } + start := int64(seq) * chunkSize + end := min(start+chunkSize, total) + eg.Go(func() error { return fetchChunk(egCtx, c, key, dst, start, end, etag) }) + } + return errors.Wrap(eg.Wait(), "parallel get") +} + +// fetchChunk opens the [start, end) range pinned to etag and writes it at start. +// An ETag change (the object was rewritten mid-download) or a short read is +// reported as an error. +func fetchChunk(ctx context.Context, c RangeReader, key Key, dst io.WriterAt, start, end int64, etag string) error { + rc, headers, err := c.Open(ctx, key, Range(start, end), IfRange(etag)) + if err != nil { + return errors.Errorf("open range %d-%d: %w", start, end, err) + } + if got := headers.Get(ETagKey); got != etag { + return errors.Join( + errors.Errorf("object changed during read at offset %d: etag %q != %q", start, got, etag), + rc.Close(), + ) + } + return writeChunkAt(dst, start, end-start, rc) +} + +// writeChunkAt streams src into dst at off and closes src. It fails if fewer +// than want bytes arrive; a negative want skips that check (total size unknown). +func writeChunkAt(dst io.WriterAt, off, want int64, src io.ReadCloser) error { + n, copyErr := io.Copy(io.NewOffsetWriter(dst, off), src) + if err := errors.Join(copyErr, src.Close()); err != nil { + return errors.Errorf("write chunk at offset %d: %w", off, err) + } + if want >= 0 && n != want { + return errors.Errorf("short chunk at offset %d: wrote %d of %d bytes", off, n, want) + } + return nil +} + +// parseContentRangeTotal extracts the total size from a Content-Range value of +// the form "bytes start-end/total". It returns ok=false when the header is +// absent or unparseable. +func parseContentRangeTotal(contentRange string) (total int64, ok bool) { + _, size, found := strings.Cut(contentRange, "/") + if !found { + return 0, false + } + total, err := strconv.ParseInt(size, 10, 64) + if err != nil { + return 0, false + } + return total, true +} diff --git a/client/parallel_get_test.go b/client/parallel_get_test.go new file mode 100644 index 00000000..df17c3a3 --- /dev/null +++ b/client/parallel_get_test.go @@ -0,0 +1,120 @@ +package client_test + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "strconv" + "sync" + "testing" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/client" +) + +// Compile-time assertion that the concrete Client satisfies the narrow +// interface ParallelGet drives. +var _ client.RangeReader = (*client.Client)(nil) + +// bufferAt is an in-memory io.WriterAt that extends like a file, zero-filling +// any gap, so tests can assert reassembly without touching disk. +type bufferAt struct { + mu sync.Mutex + buf []byte +} + +func (b *bufferAt) WriteAt(p []byte, off int64) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + if end := int(off) + len(p); end > len(b.buf) { + b.buf = append(b.buf, make([]byte, end-len(b.buf))...) + } + copy(b.buf[off:], p) + return len(p), nil +} + +// rangeFlipReader serves correct byte ranges but reports a different ETag for +// any chunk past the first, simulating an object rewritten mid-download. +type rangeFlipReader struct { + data []byte + firstETag string + restETag string +} + +func (f *rangeFlipReader) Open(_ context.Context, _ client.Key, opts ...client.RequestOption) (io.ReadCloser, http.Header, error) { + size := int64(len(f.data)) + start, length, outcome := client.NewRequestOptions(opts...).ResolveRange(size, f.firstETag) + headers := http.Header{} + if outcome == client.RangeNotSatisfiable { + headers.Set("Content-Range", fmt.Sprintf("bytes */%d", size)) + return nil, headers, client.ErrRangeNotSatisfiable + } + + etag := f.firstETag + if start > 0 { + etag = f.restETag + } + headers.Set(client.ETagKey, etag) + headers.Set("Content-Length", strconv.FormatInt(length, 10)) + if outcome == client.RangePartial { + headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, size)) + } + return io.NopCloser(bytes.NewReader(f.data[start : start+length])), headers, nil +} + +func TestParallelGetETagMismatch(t *testing.T) { + c := &rangeFlipReader{data: make([]byte, 1000), firstETag: `"v1"`, restETag: `"v2"`} + var dst bufferAt + err := client.ParallelGet(context.Background(), c, client.NewKey("k"), &dst, 100, 4) + assert.Error(t, err) + assert.Contains(t, err.Error(), "object changed during read") +} + +// noETagReader serves byte ranges but never sets an ETag, modelling a legacy +// entry or a RangeReader implementation that omits it. +type noETagReader struct { + data []byte +} + +func (n *noETagReader) Open(_ context.Context, _ client.Key, opts ...client.RequestOption) (io.ReadCloser, http.Header, error) { + size := int64(len(n.data)) + start, length, outcome := client.NewRequestOptions(opts...).ResolveRange(size, "") + headers := http.Header{} + if outcome == client.RangeNotSatisfiable { + headers.Set("Content-Range", fmt.Sprintf("bytes */%d", size)) + return nil, headers, client.ErrRangeNotSatisfiable + } + headers.Set("Content-Length", strconv.FormatInt(length, 10)) + if outcome == client.RangePartial { + headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, size)) + } + return io.NopCloser(bytes.NewReader(n.data[start : start+length])), headers, nil +} + +func TestParallelGetNoETagMultiChunk(t *testing.T) { + // A multi-chunk object with no ETag can't be pinned, so it falls back to a + // single full read (backwards compatible with objects stored before ETags). + data := make([]byte, 1000) + for i := range data { + data[i] = byte(i % 251) + } + c := &noETagReader{data: data} + var dst bufferAt + err := client.ParallelGet(context.Background(), c, client.NewKey("k"), &dst, 100, 4) + assert.NoError(t, err) + assert.Equal(t, data, dst.buf) +} + +func TestParallelGetNoETagSingleChunk(t *testing.T) { + // A no-ETag object delivered entirely by the discovery request is a single + // revision, so it succeeds without pinning. + data := []byte("0123456789") + c := &noETagReader{data: data} + var dst bufferAt + err := client.ParallelGet(context.Background(), c, client.NewKey("k"), &dst, 100, 4) + assert.NoError(t, err) + assert.Equal(t, data, dst.buf) +} diff --git a/client/preconditions.go b/client/preconditions.go index 53ada03a..79f9f066 100644 --- a/client/preconditions.go +++ b/client/preconditions.go @@ -9,6 +9,9 @@ import ( "github.com/alecthomas/errors" ) +// ETagKey is the HTTP header key used to store the ETag. +const ETagKey = "ETag" + // ErrNotModified is returned when an If-None-Match precondition is satisfied, // indicating the resource has not changed since the supplied ETag. Over HTTP // this corresponds to 304 Not Modified. diff --git a/internal/cache/etag.go b/internal/cache/etag.go index f6a8366a..d689e387 100644 --- a/internal/cache/etag.go +++ b/internal/cache/etag.go @@ -5,10 +5,12 @@ import ( "encoding/hex" "hash" "net/http" + + "github.com/block/cachew/client" ) // ETagKey is the HTTP header key used to store the ETag. -const ETagKey = "ETag" +const ETagKey = client.ETagKey // etagWriter computes a SHA256 hash of all written data. After all data has // been written, call SetETag to populate the ETag header. diff --git a/internal/cache/parallel_get.go b/internal/cache/parallel_get.go index c2cacacc..55458d07 100644 --- a/internal/cache/parallel_get.go +++ b/internal/cache/parallel_get.go @@ -3,138 +3,15 @@ package cache import ( "context" "io" - "strconv" - "strings" "github.com/alecthomas/errors" - "golang.org/x/sync/errgroup" + + "github.com/block/cachew/client" ) -// ParallelGet downloads an object from any Range-capable Cache into dst, fetching -// it in chunkSize-byte chunks concurrently (up to concurrency requests in -// flight) and writing each chunk at its offset via dst.WriteAt. Latency-bound -// backends such as a remote cache can saturate bandwidth with overlapping reads. -// -// The first chunk is fetched with a ranged Open, whose response yields both the -// total size (from Content-Range) and the object's ETag; every remaining chunk -// is then requested with IfRange pinned to that ETag. If the object changes -// mid-download, a chunk's ETag will differ and ParallelGet returns an error -// rather than splicing bytes from two revisions. A missing or truncated chunk -// is likewise reported as an error, so a partially written dst must be discarded -// by the caller on failure. An object with no ETag to pin to (e.g. one stored -// before ETags were recorded) cannot be kept revision-safe across chunks, so it -// falls back to a single full read instead of parallelising. -// -// dst is written via concurrent WriteAt calls at non-overlapping offsets; the -// caller owns dst's lifecycle (open, close, cleanup) and need not pre-size it, -// as WriteAt extends the destination. +// ParallelGet downloads an object from any Range-capable Cache into dst, +// fetching it in chunkSize-byte chunks concurrently. It delegates to +// [client.ParallelGet]; see that function for the full semantics. func ParallelGet(ctx context.Context, c Cache, key Key, dst io.WriterAt, chunkSize int64, concurrency int) error { - if chunkSize <= 0 { - return errors.Errorf("parallel get: chunk size must be positive, got %d", chunkSize) - } - concurrency = max(concurrency, 1) - - // Discovery: the first ranged Open delivers chunk zero and reveals the total - // size and ETag used to pin the rest. - rc, headers, err := c.Open(ctx, key, Range(0, chunkSize)) - if errors.Is(err, ErrRangeNotSatisfiable) { - return nil // Empty object: nothing to write. - } - if err != nil { - return errors.Wrap(err, "parallel get: open first chunk") - } - - etag := headers.Get(ETagKey) - total, hasRange := parseContentRangeTotal(headers.Get("Content-Range")) - - // A backend that ignored the range (no Content-Range), or an object that - // fits within the first chunk, is delivered entirely by this response: copy - // it and return, as there is nothing to parallelise. A negative want skips - // the length check when the total size is unknown. - firstLen := min(chunkSize, total) - if !hasRange { - firstLen = -1 - } - if !hasRange || total <= chunkSize { - return errors.Wrap(writeChunkAt(dst, 0, firstLen, rc), "parallel get") - } - - // Subsequent chunks are pinned to the discovery ETag via IfRange. Without a - // validator there is nothing to pin to (IfRange("") is a no-op and an empty - // ETag matches an empty ETag), so chunks could be spliced across a rewrite - // undetected. Objects stored before ETags were recorded fall here, so fall - // back to a single, revision-consistent read rather than parallelising. - if etag == "" { - if err := rc.Close(); err != nil { - return errors.Wrap(err, "parallel get: close discovery reader") - } - full, _, err := c.Open(ctx, key) - if err != nil { - return errors.Wrap(err, "parallel get: full read") - } - return errors.Wrap(writeChunkAt(dst, 0, total, full), "parallel get") - } - - // Multiple chunks: copy the already-open first chunk concurrently with the - // rest rather than blocking on it here. The first goroutine is scheduled - // before the limit can be reached, so it never stalls holding an open body. - numChunks := int((total + chunkSize - 1) / chunkSize) - eg, egCtx := errgroup.WithContext(ctx) - eg.SetLimit(concurrency) - eg.Go(func() error { return writeChunkAt(dst, 0, firstLen, rc) }) - for seq := 1; seq < numChunks; seq++ { - // Stop scheduling once a chunk has failed and cancelled the group. - if egCtx.Err() != nil { - break - } - start := int64(seq) * chunkSize - end := min(start+chunkSize, total) - eg.Go(func() error { return fetchChunk(egCtx, c, key, dst, start, end, etag) }) - } - return errors.Wrap(eg.Wait(), "parallel get") -} - -// fetchChunk opens the [start, end) range pinned to etag and writes it at start. -// An ETag change (the object was rewritten mid-download) or a short read is -// reported as an error. -func fetchChunk(ctx context.Context, c Cache, key Key, dst io.WriterAt, start, end int64, etag string) error { - rc, headers, err := c.Open(ctx, key, Range(start, end), IfRange(etag)) - if err != nil { - return errors.Errorf("open range %d-%d: %w", start, end, err) - } - if got := headers.Get(ETagKey); got != etag { - return errors.Join( - errors.Errorf("object changed during read at offset %d: etag %q != %q", start, got, etag), - rc.Close(), - ) - } - return writeChunkAt(dst, start, end-start, rc) -} - -// writeChunkAt streams src into dst at off and closes src. It fails if fewer -// than want bytes arrive; a negative want skips that check (total size unknown). -func writeChunkAt(dst io.WriterAt, off, want int64, src io.ReadCloser) error { - n, copyErr := io.Copy(io.NewOffsetWriter(dst, off), src) - if err := errors.Join(copyErr, src.Close()); err != nil { - return errors.Errorf("write chunk at offset %d: %w", off, err) - } - if want >= 0 && n != want { - return errors.Errorf("short chunk at offset %d: wrote %d of %d bytes", off, n, want) - } - return nil -} - -// parseContentRangeTotal extracts the total size from a Content-Range value of -// the form "bytes start-end/total". It returns ok=false when the header is -// absent or unparseable. -func parseContentRangeTotal(contentRange string) (total int64, ok bool) { - _, size, found := strings.Cut(contentRange, "/") - if !found { - return 0, false - } - total, err := strconv.ParseInt(size, 10, 64) - if err != nil { - return 0, false - } - return total, true + return errors.WithStack(client.ParallelGet(ctx, c, key, dst, chunkSize, concurrency)) } diff --git a/internal/cache/parallel_get_test.go b/internal/cache/parallel_get_test.go index b1a280d5..9f0b5a4f 100644 --- a/internal/cache/parallel_get_test.go +++ b/internal/cache/parallel_get_test.go @@ -1,14 +1,10 @@ package cache_test import ( - "bytes" "context" - "fmt" "io" "log/slog" - "net/http" "os" - "strconv" "sync" "testing" "time" @@ -99,88 +95,3 @@ func TestParallelGetNotFound(t *testing.T) { err = cache.ParallelGet(ctx, c, cache.NewKey("missing"), &dst, 100, 4) assert.IsError(t, err, os.ErrNotExist) } - -// rangeFlipCache serves correct byte ranges but reports a different ETag for any -// chunk past the first, simulating an object rewritten mid-download. -type rangeFlipCache struct { - cache.Cache // embedded; only Open is exercised by ParallelGet - data []byte - firstETag string - restETag string -} - -func (f *rangeFlipCache) Open(_ context.Context, _ cache.Key, opts ...cache.Option) (io.ReadCloser, http.Header, error) { - size := int64(len(f.data)) - start, length, outcome := cache.NewRequestOptions(opts...).ResolveRange(size, f.firstETag) - headers := http.Header{} - if outcome == cache.RangeNotSatisfiable { - headers.Set("Content-Range", fmt.Sprintf("bytes */%d", size)) - return nil, headers, cache.ErrRangeNotSatisfiable - } - - etag := f.firstETag - if start > 0 { - etag = f.restETag - } - headers.Set("ETag", etag) - headers.Set("Content-Length", strconv.FormatInt(length, 10)) - if outcome == cache.RangePartial { - headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, size)) - } - return io.NopCloser(bytes.NewReader(f.data[start : start+length])), headers, nil -} - -func TestParallelGetETagMismatch(t *testing.T) { - c := &rangeFlipCache{data: make([]byte, 1000), firstETag: `"v1"`, restETag: `"v2"`} - var dst bufferAt - err := cache.ParallelGet(context.Background(), c, cache.NewKey("k"), &dst, 100, 4) - assert.Error(t, err) - assert.Contains(t, err.Error(), "object changed during read") -} - -// noETagCache serves byte ranges but never sets an ETag, modelling a legacy -// entry or a Cache implementation that omits it. -type noETagCache struct { - cache.Cache // embedded; only Open is exercised - data []byte -} - -func (n *noETagCache) Open(_ context.Context, _ cache.Key, opts ...cache.Option) (io.ReadCloser, http.Header, error) { - size := int64(len(n.data)) - start, length, outcome := cache.NewRequestOptions(opts...).ResolveRange(size, "") - headers := http.Header{} - if outcome == cache.RangeNotSatisfiable { - headers.Set("Content-Range", fmt.Sprintf("bytes */%d", size)) - return nil, headers, cache.ErrRangeNotSatisfiable - } - headers.Set("Content-Length", strconv.FormatInt(length, 10)) - if outcome == cache.RangePartial { - headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, size)) - } - return io.NopCloser(bytes.NewReader(n.data[start : start+length])), headers, nil -} - -func TestParallelGetNoETagMultiChunk(t *testing.T) { - // A multi-chunk object with no ETag can't be pinned, so it falls back to a - // single full read (backwards compatible with objects stored before ETags). - data := make([]byte, 1000) - for i := range data { - data[i] = byte(i % 251) - } - c := &noETagCache{data: data} - var dst bufferAt - err := cache.ParallelGet(context.Background(), c, cache.NewKey("k"), &dst, 100, 4) - assert.NoError(t, err) - assert.Equal(t, data, dst.buf) -} - -func TestParallelGetNoETagSingleChunk(t *testing.T) { - // A no-ETag object delivered entirely by the discovery request is a single - // revision, so it succeeds without pinning. - data := []byte("0123456789") - c := &noETagCache{data: data} - var dst bufferAt - err := cache.ParallelGet(context.Background(), c, cache.NewKey("k"), &dst, 100, 4) - assert.NoError(t, err) - assert.Equal(t, data, dst.buf) -}