Native and remote streaming improvements

pkg/archive/archive_exploded.go

@@ -31,13 +31,17 @@ func (e explodedArchiveEntry) CRC32Checksum() *uint32 {
 }
 
 func (e explodedArchiveEntry) CompressedAs(compressionMethod CompressionMethod) bool {
-	return false
+	// Exploded-archive entries are plain files, i.e. stored uncompressed.
+	return compressionMethod == CompressionMethodStore
 }
 
 func (e explodedArchiveEntry) Read(start int64, end int64) ([]byte, error) {
 	if end < start {
 		return nil, errors.New("range not satisfiable")
 	}
+	if start < 0 {
+		start = 0
+	}
 	f, err := os.Open(filepath.Join(e.dir, e.filepath))
 	if err != nil {
 		return nil, err
@@ -70,6 +74,9 @@ func (e explodedArchiveEntry) Stream(w io.Writer, start int64, end int64) (int64
 	if end < start {
 		return -1, errors.New("range not satisfiable")
 	}
+	if start < 0 {
+		start = 0
+	}
 	f, err := os.Open(filepath.Join(e.dir, e.filepath))
 	if err != nil {
 		return -1, err

pkg/archive/archive_zip.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"io/fs"
 	"math"
+	"os"
 	"path"
 	"sync"
 
@@ -19,10 +20,51 @@ type gozipArchiveEntry struct {
 	file          *zip.File
 	minimizeReads bool
 
+	// path is the filesystem path of the archive when it was opened from a
+	// local file, or empty otherwise. It unlocks streaming an entry's raw
+	// bytes straight from a file handle (kernel sendfile fast path).
+	path string
+
 	gi zran.Index
 	gm sync.Mutex
 }
 
+// zipFlagEncrypted is the general-purpose bit flag marking an encrypted entry,
+// whose raw bytes are not the actual content.
+const zipFlagEncrypted = 0x1
+
+// hasRawFileAccess reports whether the entry's raw bytes can be served
+// directly from the archive's file on disk.
+func (e *gozipArchiveEntry) hasRawFileAccess() bool {
+	return e.path != "" && e.file.Flags&zipFlagEncrypted == 0
+}
+
+// streamRawFromFile copies length raw bytes of the entry, skipping the first
+// start bytes, straight from a fresh handle on the archive file. The source
+// reaching w is a bare *os.File (wrapped only by io.CopyN's *io.LimitedReader,
+// which the runtime unwraps), so the kernel sendfile fast path is preserved
+// when w is a network connection. handled is false when the fast path could
+// not be attempted and the caller should fall back before writing anything.
+func (e *gozipArchiveEntry) streamRawFromFile(w io.Writer, start int64, length int64) (n int64, err error, handled bool) {
+	offset, err := e.file.DataOffset()
+	if err != nil {
+		return 0, nil, false
+	}
+	f, err := os.Open(e.path)
+	if err != nil {
+		return 0, nil, false
+	}
+	defer f.Close()
+	if _, err := f.Seek(offset+start, io.SeekStart); err != nil {
+		return 0, nil, false
+	}
+	n, err = io.CopyN(w, f, length)
+	if err == io.EOF {
+		err = nil
+	}
+	return n, err, true
+}
+
 func (e *gozipArchiveEntry) Path() string {
 	return path.Clean(e.file.Name)
 }
@@ -44,10 +86,14 @@ func (e *gozipArchiveEntry) CRC32Checksum() *uint32 {
 }
 
 func (e *gozipArchiveEntry) CompressedAs(compressionMethod CompressionMethod) bool {
-	if compressionMethod != CompressionMethodDeflate {
+	switch compressionMethod {
+	case CompressionMethodDeflate:
+		return e.file.Method == zip.Deflate
+	case CompressionMethodStore:
+		return e.file.Method == zip.Store
+	default:
 		return false
 	}
-	return e.file.Method == zip.Deflate
 }
 
 // This is a special mode to minimize the number of reads from the underlying reader.
@@ -62,6 +108,9 @@ func (e *gozipArchiveEntry) Read(start int64, end int64) ([]byte, error) {
 	if end < start {
 		return nil, errors.New("range not satisfiable")
 	}
+	if start < 0 {
+		start = 0
+	}
 
 	minimizeReads := e.couldMinimizeReads()
 
@@ -72,6 +121,15 @@ func (e *gozipArchiveEntry) Read(start int64, end int64) ([]byte, error) {
 		if err != nil {
 			return nil, err
 		}
+	} else if e.CompressedLength() == 0 && e.file.Flags&zipFlagEncrypted == 0 {
+		// Raw bytes == content for stored entries; OpenRaw skips the CRC32
+		// pass and returns a seekable reader, so a ranged read seeks to the
+		// range start instead of reading and discarding the prefix — which
+		// for a remote archive would download the whole prefix.
+		f, err = e.file.OpenRaw()
+		if err != nil {
+			return nil, err
+		}
 	} else {
 		rc, err := e.file.Open()
 		if err != nil {
@@ -169,6 +227,29 @@ func (e *gozipArchiveEntry) Stream(w io.Writer, start int64, end int64) (int64,
 	if end < start {
 		return -1, errors.New("range not satisfiable")
 	}
+	if start < 0 {
+		start = 0
+	}
+
+	// For stored entries the raw bytes are the content, so they can be
+	// streamed directly from the archive file on disk, bypassing both the
+	// CRC32 pass of zip's checksum reader and userspace copy loops.
+	if e.CompressedLength() == 0 && e.hasRawFileAccess() {
+		size := int64(e.file.UncompressedSize64)
+		length := size
+		if !(start == 0 && end == 0) {
+			if start >= size {
+				return 0, nil
+			}
+			length = end - start + 1
+			if length > size-start {
+				length = size - start
+			}
+		}
+		if n, err, handled := e.streamRawFromFile(w, start, length); handled {
+			return n, err
+		}
+	}
 
 	minimizeReads := e.couldMinimizeReads() && start == 0 && end == 0
 
@@ -179,6 +260,13 @@ func (e *gozipArchiveEntry) Stream(w io.Writer, start int64, end int64) (int64,
 		if err != nil {
 			return -1, err
 		}
+	} else if e.CompressedLength() == 0 && e.file.Flags&zipFlagEncrypted == 0 {
+		// Raw bytes == content for stored entries; OpenRaw skips the CRC32
+		// pass and returns a seekable reader for cheap range starts.
+		f, err = e.file.OpenRaw()
+		if err != nil {
+			return -1, err
+		}
 	} else {
 		rc, err := e.file.Open()
 		if err != nil {
@@ -200,26 +288,101 @@ func (e *gozipArchiveEntry) Stream(w io.Writer, start int64, end int64) (int64,
 	}
 
 	if start == 0 && end == 0 {
-		return io.Copy(w, f)
+		n, err := copyRange(w, f, int64(e.file.UncompressedSize64))
+		if err == io.EOF {
+			err = nil
+		}
+		return n, err
 	}
 	if start > 0 {
-		n, err := io.CopyN(io.Discard, f, start)
-		if err != nil {
+		if skr, ok := f.(io.Seeker); ok {
+			if _, err := skr.Seek(start, io.SeekStart); err != nil {
+				return -1, err
+			}
+		} else if n, err := io.CopyN(io.Discard, f, start); err != nil {
 			return n, err
 		}
 	}
-	n, err := io.CopyN(w, f, end-start+1)
+	n, err := copyRange(w, f, end-start+1)
 	if n > 0 && err == io.EOF {
 		// Not EOF error if some data was read
 		err = nil
 	}
 	return n, err
 }
 
+// streamCopyBufferSize is the buffer size used when copying entry bytes to a
+// writer. The underlying archive is read at this granularity, and for a remote
+// archive each read can become its own range request — io.Copy's default 32KB
+// buffer would turn a large stream into one remote request per 32KB. It is
+// deliberately larger than the default remote range-cache threshold (1 MiB) so
+// that streamed media blocks are not needlessly copied into the range cache.
+const streamCopyBufferSize = 2 << 20 // 2 MiB
+
+// copyBufferSize bounds the copy buffer to the number of bytes actually being
+// copied, so streaming a small entry doesn't allocate the full buffer.
+func copyBufferSize(length int64) int {
+	if length <= 0 {
+		return 1
+	}
+	if length < streamCopyBufferSize {
+		return int(length)
+	}
+	return streamCopyBufferSize
+}
+
+// copyRange copies exactly length bytes from r to w using a full-sized read
+// per iteration, following io.CopyN semantics (io.EOF when r ends early).
+//
+// It deliberately avoids io.Copy/io.CopyBuffer: those delegate to the
+// destination's ReaderFrom when available (an http.ResponseWriter does), which
+// reads the source in its own small chunks — and every read of a remote
+// archive can be its own range request. io.ReadFull guarantees each iteration
+// asks the underlying reader for the whole buffer at once.
+func copyRange(w io.Writer, r io.Reader, length int64) (int64, error) {
+	if length <= 0 {
+		return 0, nil
+	}
+	buf := make([]byte, copyBufferSize(length))
+	var written int64
+	for written < length {
+		n := int64(len(buf))
+		if rem := length - written; rem < n {
+			n = rem
+		}
+		nr, rerr := io.ReadFull(r, buf[:n])
+		if nr > 0 {
+			nw, werr := w.Write(buf[:nr])
+			written += int64(nw)
+			if werr != nil {
+				return written, werr
+			}
+			if int64(nw) < int64(nr) {
+				return written, io.ErrShortWrite
+			}
+		}
+		if rerr == io.EOF || rerr == io.ErrUnexpectedEOF {
+			break // Source exhausted before the requested length
+		}
+		if rerr != nil {
+			return written, rerr
+		}
+	}
+	if written < length {
+		return written, io.EOF
+	}
+	return written, nil
+}
+
 func (e *gozipArchiveEntry) StreamCompressed(w io.Writer) (int64, error) {
 	if e.file.Method != zip.Deflate {
 		return -1, errors.New("not a compressed resource")
 	}
+	if e.hasRawFileAccess() {
+		if n, err, handled := e.streamRawFromFile(w, 0, int64(e.file.CompressedSize64)); handled {
+			return n, err
+		}
+	}
 	f, err := e.file.OpenRaw()
 	if err != nil {
 		return -1, err
@@ -235,9 +398,13 @@ func (e *gozipArchiveEntry) StreamCompressedGzip(w io.Writer) (int64, error) {
 	if e.file.UncompressedSize64 > math.MaxUint32 {
 		return -1, errors.New("uncompressed size > 2^32 too large for GZIP")
 	}
-	f, err := e.file.OpenRaw()
-	if err != nil {
-		return -1, err
+	var f io.Reader
+	if !e.hasRawFileAccess() {
+		var err error
+		f, err = e.file.OpenRaw()
+		if err != nil {
+			return -1, err
+		}
 	}
 
 	// Header
@@ -249,7 +416,23 @@ func (e *gozipArchiveEntry) StreamCompressedGzip(w io.Writer) (int64, error) {
 		return -1, errors.Wrap(err, "failed to write GZIP header")
 	}
 
-	nn, err := io.Copy(w, f)
+	var nn int64
+	if f == nil {
+		// The deflate body is copied straight from the archive file so the
+		// kernel sendfile fast path applies between header and trailer.
+		var handled bool
+		nn, err, handled = e.streamRawFromFile(w, 0, int64(e.file.CompressedSize64))
+		if !handled {
+			var oerr error
+			f, oerr = e.file.OpenRaw()
+			if oerr != nil {
+				return int64(n), oerr
+			}
+			nn, err = io.Copy(w, f)
+		}
+	} else {
+		nn, err = io.Copy(w, f)
+	}
 	if err != nil {
 		return int64(n), errors.Wrap(err, "failed copying deflated bytes")
 	}
@@ -322,6 +505,10 @@ type gozipArchive struct {
 	closer        func() error
 	minimizeReads bool
 
+	// path is the filesystem path of the archive when it was opened from a
+	// local file, or empty otherwise. See gozipArchiveEntry.path.
+	path string
+
 	// nameIndex maps each entry's cleaned path to its *zip.File, giving O(1)
 	// lookups instead of a linear scan. It's built once, lazily, on the first
 	// Entry call that misses the wrapper cache.
@@ -356,6 +543,7 @@ func (a *gozipArchive) wrap(cleanPath string, f *zip.File) *gozipArchiveEntry {
 	entry := &gozipArchiveEntry{
 		file:          f,
 		minimizeReads: a.minimizeReads,
+		path:          a.path,
 	}
 	actual, _ := a.cachedEntries.LoadOrStore(cleanPath, entry)
 	return actual.(*gozipArchiveEntry)
@@ -419,7 +607,11 @@ func (e gozipArchiveFactory) Open(filepath string, password string) (Archive, er
 	if err != nil {
 		return nil, err
 	}
-	return NewGoZIPArchive(&rc.Reader, rc.Close, false), nil
+	return &gozipArchive{
+		zip:    &rc.Reader,
+		closer: rc.Close,
+		path:   filepath,
+	}, nil
 }
 
 // OpenBytes implements ArchiveFactory