diff --git a/internal/cache/disk_test.go b/internal/cache/disk_test.go index 3c57e1cc..09864a4e 100644 --- a/internal/cache/disk_test.go +++ b/internal/cache/disk_test.go @@ -9,6 +9,7 @@ import ( "net/http" "os" "sync" + "syscall" "testing" "time" @@ -139,3 +140,44 @@ func TestDiskOpenRevisionAtomic(t *testing.T) { }() wg.Wait() } + +// fileSinkWriter records what io.Copy hands to ReadFrom, as net/http does when +// it passes a body to the connection's sendfile path. +type fileSinkWriter struct { + bytes.Buffer + src io.Reader +} + +func (w *fileSinkWriter) ReadFrom(r io.Reader) (int64, error) { + w.src = r + return w.Buffer.ReadFrom(r) +} + +// TestDiskRangeReachesSendfile verifies a ranged disk read reaches ReadFrom as +// an *io.LimitedReader over the file, the only shape net's sendfile accepts. +func TestDiskRangeReachesSendfile(t *testing.T) { + _, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelError}) + c, err := cache.NewDisk(ctx, cache.DiskConfig{Root: t.TempDir(), MaxTTL: time.Hour}) + assert.NoError(t, err) + defer c.Close() + + key := cache.NewKey("range-sendfile") + w, err := c.Create(ctx, key, nil, time.Hour) + assert.NoError(t, err) + _, err = w.Write([]byte("0123456789abcdef")) + assert.NoError(t, err) + assert.NoError(t, w.Close()) + + r, _, err := c.Open(ctx, key, cache.Range(4, 10)) + assert.NoError(t, err) + + var sink fileSinkWriter + _, err = io.Copy(&sink, r) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + assert.Equal(t, "456789", sink.String()) + limited, ok := sink.src.(*io.LimitedReader) + assert.True(t, ok, "ReadFrom got %T", sink.src) + _, ok = limited.R.(syscall.Conn) + assert.True(t, ok, "LimitedReader wraps %T", limited.R) +} diff --git a/internal/cache/range.go b/internal/cache/range.go index faea96c9..76df3dea 100644 --- a/internal/cache/range.go +++ b/internal/cache/range.go @@ -48,3 +48,11 @@ func newLimitedReadCloser(rc io.ReadCloser, length int64) io.ReadCloser { } func (l limitedReadCloser) Close() error { return errors.Wrap(l.closer.Close(), "close range reader") } + +// WriteTo hands w the underlying *io.LimitedReader rather than this wrapper. +// net/http passes it to the connection's ReadFrom, which recognises a +// LimitedReader over an *os.File and serves a disk range with sendfile(2). +func (l limitedReadCloser) WriteTo(w io.Writer) (int64, error) { + n, err := io.Copy(w, l.Reader) + return n, errors.Wrap(err, "copy range") +} diff --git a/internal/strategy/git/snapshot.go b/internal/strategy/git/snapshot.go index cd96ac41..f4325b5d 100644 --- a/internal/strategy/git/snapshot.go +++ b/internal/strategy/git/snapshot.go @@ -689,37 +689,6 @@ func (s *Strategy) serveSnapshotHead(ctx context.Context, w http.ResponseWriter, } } -func (s *Strategy) streamSnapshotArtifact(_ context.Context, w http.ResponseWriter, r *http.Request, reader io.ReadCloser, headers http.Header) error { - for key, values := range headers { - for _, value := range values { - w.Header().Add(key, value) - } - } - if _, err := serveReaderFast(w, r, reader); err != nil { - return errors.Wrap(err, "streaming artifact") - } - return nil -} - -// serveReaderFast serves the content from reader using the most efficient method -// available. When reader is an *os.File, it uses http.ServeContent which enables -// sendfile(2) zero-copy I/O and automatic Content-Length/Range support. For other -// reader types it falls back to io.Copy. Returns bytes served for metrics. -func serveReaderFast(w http.ResponseWriter, r *http.Request, reader io.Reader) (int64, error) { - if f, ok := reader.(*os.File); ok { - info, err := f.Stat() - if err != nil { - return 0, errors.Wrap(err, "stat file for serving") - } - // http.ServeContent handles Content-Length, Range requests, and uses - // sendfile(2) for zero-copy transfer from file to socket. - http.ServeContent(w, r, "", time.Time{}, f) - return info.Size(), nil - } - n, err := io.Copy(w, reader) - return n, errors.Wrap(err, "copy to response") -} - func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, host, pathValue string) { //nolint:funlen start := time.Now() repoPath := ExtractRepoPath(strings.TrimSuffix(pathValue, "/snapshot.bundle")) @@ -1687,19 +1656,20 @@ func (s *Strategy) handleLFSSnapshotRequest(w http.ResponseWriter, r *http.Reque upstreamURL := "https://" + host + "/" + repoPath cacheKey := lfsSnapshotCacheKey(upstreamURL) - // Try cache first so we can serve even when the mirror isn't ready (cold start). - reader, headers, err := s.cache.Open(ctx, cacheKey) - if err != nil && !errors.Is(err, os.ErrNotExist) { - logger.ErrorContext(ctx, "Failed to open LFS snapshot from cache", "upstream", upstreamURL, "error", err) - http.Error(w, "Internal server error", http.StatusInternalServerError) + // Try cache first so we can serve even when the mirror isn't ready (cold + // start). Range and the validators go to the cache, as for the working-tree + // snapshot, so an interrupted download can resume pinned to its ETag. + reader, headers, err := s.cache.Open(ctx, cacheKey, httputil.ConditionalOptions(r)...) + handled, _, serveErr := httputil.ServeCacheHit(w, headers, reader, err) + if serveErr != nil { + logger.ErrorContext(ctx, "Failed to stream LFS snapshot", "upstream", upstreamURL, "error", serveErr) + } + if handled { return } - if reader != nil { - defer reader.Close() - logger.DebugContext(ctx, "Serving cached LFS snapshot", "upstream", upstreamURL) - if err := s.streamSnapshotArtifact(ctx, w, r, reader, headers); err != nil { - logger.ErrorContext(ctx, "Failed to stream LFS snapshot", "upstream", upstreamURL, "error", err) - } + if !errors.Is(err, os.ErrNotExist) { + logger.ErrorContext(ctx, "Failed to open LFS snapshot from cache", "upstream", upstreamURL, "error", err) + http.Error(w, "Internal server error", http.StatusInternalServerError) return } diff --git a/internal/strategy/git/snapshot_test.go b/internal/strategy/git/snapshot_test.go index ce1ec823..7df9d26d 100644 --- a/internal/strategy/git/snapshot_test.go +++ b/internal/strategy/git/snapshot_test.go @@ -1354,6 +1354,60 @@ func TestSnapshotGetHonorsRange(t *testing.T) { assert.Equal(t, http.StatusPreconditionFailed, rangeWithStaleIfMatch.Code) } +func TestLFSSnapshotGetHonorsRangeAndIfMatch(t *testing.T) { + _, ctx := logging.Configure(context.Background(), logging.Config{}) + upstreamURL := "https://github.com/org/repo" + body := []byte("0123456789abcdef") + + // A non-*os.File reader, so the range must be resolved by the cache. + memCache, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + w, err := memCache.Create(ctx, cache.NewKey(upstreamURL+".lfs-snapshot"), http.Header{"Content-Type": {"application/zstd"}}, time.Hour) + assert.NoError(t, err) + _, err = w.Write(body) + assert.NoError(t, err) + assert.NoError(t, w.Close()) + + mux := newTestMux() + cm := gitclone.NewManagerProvider(ctx, gitclone.Config{MirrorRoot: t.TempDir()}, nil) + _, err = git.New(ctx, git.Config{}, newTestScheduler(ctx, t), memCache, mux, cm, func() (*githubapp.TokenManager, error) { return nil, nil }, newTestSpools(t)) //nolint:nilnil + assert.NoError(t, err) + handler := mux.handlers["GET /git/{host}/{path...}"] + assert.NotZero(t, handler) + + get := func(headers map[string]string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, "/git/github.com/org/repo/lfs-snapshot.tar.zst", nil) + req = req.WithContext(ctx) + req.SetPathValue("host", "github.com") + req.SetPathValue("path", "org/repo/lfs-snapshot.tar.zst") + for k, v := range headers { + req.Header.Set(k, v) + } + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + return rec + } + + full := get(nil) + assert.Equal(t, http.StatusOK, full.Code) + assert.Equal(t, body, full.Body.Bytes()) + assert.Equal(t, "bytes", full.Header().Get("Accept-Ranges")) + etag := full.Header().Get("ETag") + assert.NotZero(t, etag) + + resumed := get(map[string]string{"Range": "bytes=10-", "If-Match": etag}) + assert.Equal(t, http.StatusPartialContent, resumed.Code) + assert.Equal(t, "bytes 10-15/16", resumed.Header().Get("Content-Range")) + assert.Equal(t, body[10:], resumed.Body.Bytes()) + + regenerated := get(map[string]string{"Range": "bytes=10-", "If-Match": `"stale-etag"`}) + assert.Equal(t, http.StatusPreconditionFailed, regenerated.Code) + assert.Equal(t, 0, regenerated.Body.Len()) + + notModified := get(map[string]string{"If-None-Match": etag}) + assert.Equal(t, http.StatusNotModified, notModified.Code) +} + // createUpstreamAndMirror creates a working upstream repo with one commit and // a mirror clone of it, returning the upstream work dir so tests can add // commits upstream after the mirror is created.