More improvements in streaming caches for M4B cases

pkg/fetcher/fetcher_gcs.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"path"
 	"strings"
+	"sync"
 
 	"cloud.google.com/go/storage"
 	"github.com/readium/go-toolkit/pkg/manifest"
@@ -22,6 +23,12 @@ type GCSFetcher struct {
 	handle *storage.ObjectHandle
 
 	cachedLinks manifest.LinkList
+
+	// attrsCache shares object metadata (object name -> *storage.ObjectAttrs)
+	// across the resources created by Get, so serving many requests for the
+	// same object — every one of which needs the length — performs a single
+	// attributes call over the fetcher's lifetime instead of one per request.
+	attrsCache sync.Map
 }
 
 func NewGCSFetcher(href string, client *storage.Client, handle *storage.ObjectHandle) *GCSFetcher {
@@ -136,8 +143,9 @@ func (f *GCSFetcher) Get(ctx context.Context, link manifest.Link) Resource {
 		// must not let an incoming request reach objects elsewhere in the bucket.
 		if keyWithinRoot(f.handle.ObjectName(), resourceFile) {
 			return &gcsResource{
-				handle: f.client.Bucket(f.handle.BucketName()).Object(resourceFile),
-				link:   link,
+				handle:     f.client.Bucket(f.handle.BucketName()).Object(resourceFile),
+				link:       link,
+				attrsCache: &f.attrsCache,
 			}
 		}
 	}
@@ -154,6 +162,7 @@ type gcsResource struct {
 	link        manifest.Link
 	handle      *storage.ObjectHandle
 	cachedAttrs *storage.ObjectAttrs
+	attrsCache  *sync.Map // Optional fetcher-shared metadata cache, see GCSFetcher
 }
 
 // Link implements Resource
@@ -177,12 +186,20 @@ func (r *gcsResource) File() string {
 }
 
 func (r *gcsResource) attrs(ctx context.Context) (*storage.ObjectAttrs, *ResourceError) {
+	if r.cachedAttrs == nil && r.attrsCache != nil {
+		if v, ok := r.attrsCache.Load(r.handle.ObjectName()); ok {
+			r.cachedAttrs = v.(*storage.ObjectAttrs)
+		}
+	}
 	if r.cachedAttrs == nil {
 		head, err := r.handle.Attrs(ctx)
 		if err != nil {
 			return nil, gcsErrorToException(err)
 		}
 		r.cachedAttrs = head
+		if r.attrsCache != nil {
+			r.attrsCache.Store(r.handle.ObjectName(), head)
+		}
 	}
 	return r.cachedAttrs, nil
 }

pkg/fetcher/fetcher_http.go

@@ -8,6 +8,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/pkg/errors"
 	"github.com/readium/go-toolkit/pkg/manifest"
@@ -19,6 +20,12 @@ type HTTPFetcher struct {
 	href   string
 	client *http.Client
 	url    url.AbsoluteURL
+
+	// sizes shares resource sizes (resolved URL -> int64) across the resources
+	// created by Get, so serving many requests for the same resource — every
+	// one of which needs the length — performs a single HEAD request over the
+	// fetcher's lifetime instead of one per request.
+	sizes sync.Map
 }
 
 func NewHTTPFetcher(href string, client *http.Client, url url.AbsoluteURL) *HTTPFetcher {
@@ -75,6 +82,7 @@ func (f *HTTPFetcher) Get(ctx context.Context, link manifest.Link) Resource {
 					link:   link,
 					client: f.client,
 					url:    resolved,
+					sizes:  &f.sizes,
 				}
 			}
 		}
@@ -106,6 +114,7 @@ type httpResource struct {
 	url    url.AbsoluteURL
 
 	cachedSize *int64
+	sizes      *sync.Map // Optional fetcher-shared size cache, see HTTPFetcher
 }
 
 // Link implements Resource
@@ -129,6 +138,12 @@ func (r *httpResource) File() string {
 }
 
 func (r *httpResource) size(ctx context.Context) (int64, *ResourceError) {
+	if r.cachedSize == nil && r.sizes != nil {
+		if v, ok := r.sizes.Load(r.url.String()); ok {
+			length := v.(int64)
+			r.cachedSize = &length
+		}
+	}
 	if r.cachedSize == nil {
 		req, err := http.NewRequestWithContext(ctx, http.MethodHead, r.url.String(), nil)
 		if err != nil {
@@ -159,7 +174,9 @@ func (r *httpResource) size(ctx context.Context) (int64, *ResourceError) {
 			return 0, Other(errors.Wrap(err, "failed to parse Content-Length header"))
 		}
 		r.cachedSize = &length
-
+		if r.sizes != nil {
+			r.sizes.Store(r.url.String(), length)
+		}
 	}
 	return *r.cachedSize, nil
 }

pkg/fetcher/fetcher_http_test.go

@@ -95,3 +95,39 @@ func TestHTTPResourceStream(t *testing.T) {
 	assert.Equal(t, []string{"bytes=100-199"}, ranges)
 	mu.Unlock()
 }
+
+// Resources created by the same HTTPFetcher share learned sizes, so serving
+// many requests for the same file — each of which needs Length for the
+// Content-Length header — performs a single HEAD request in total.
+func TestHTTPFetcherSharesResourceSize(t *testing.T) {
+	payload := make([]byte, 1234)
+
+	var mu sync.Mutex
+	heads := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodHead {
+			mu.Lock()
+			heads++
+			mu.Unlock()
+		}
+		http.ServeContent(w, r, "file.bin", time.Time{}, bytes.NewReader(payload))
+	}))
+	defer srv.Close()
+
+	base, err := url.AbsoluteURLFromString(srv.URL + "/")
+	require.NoError(t, err)
+	f := NewHTTPFetcher("", srv.Client(), base)
+	link := manifest.Link{Href: manifest.MustNewHREFFromString("file.bin", false)}
+
+	for range 3 {
+		res := f.Get(t.Context(), link)
+		n, rerr := res.Length(t.Context())
+		require.Nil(t, rerr)
+		assert.Equal(t, int64(1234), n)
+		res.Close()
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, 1, heads, "size should be learned once and shared across resources")
+}

pkg/fetcher/fetcher_s3.go

@@ -7,6 +7,7 @@ import (
 	"path"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -24,6 +25,12 @@ type S3Fetcher struct {
 	key    string
 
 	cachedLinks manifest.LinkList
+
+	// heads shares object metadata (key -> *s3.HeadObjectOutput) across the
+	// resources created by Get, so serving many requests for the same object —
+	// every one of which needs the length — performs a single HeadObject call
+	// over the fetcher's lifetime instead of one per request.
+	heads sync.Map
 }
 
 func NewS3Fetcher(href string, client *s3.Client, bucket, key string) *S3Fetcher {
@@ -126,6 +133,7 @@ func (f *S3Fetcher) Get(ctx context.Context, link manifest.Link) Resource {
 				client: f.client,
 				bucket: f.bucket,
 				key:    resourceFile,
+				heads:  &f.heads,
 			}
 		}
 	}
@@ -145,6 +153,7 @@ type s3Resource struct {
 	key    string
 
 	cachedHead *s3.HeadObjectOutput
+	heads      *sync.Map // Optional fetcher-shared metadata cache, see S3Fetcher
 }
 
 // Link implements Resource
@@ -175,6 +184,11 @@ func (r *s3Resource) object() *s3.GetObjectInput {
 }
 
 func (r *s3Resource) head(ctx context.Context) (*s3.HeadObjectOutput, *ResourceError) {
+	if r.cachedHead == nil && r.heads != nil {
+		if v, ok := r.heads.Load(r.key); ok {
+			r.cachedHead = v.(*s3.HeadObjectOutput)
+		}
+	}
 	if r.cachedHead == nil {
 		head, err := r.client.HeadObject(ctx, &s3.HeadObjectInput{
 			Bucket: &r.bucket,
@@ -184,6 +198,9 @@ func (r *s3Resource) head(ctx context.Context) (*s3.HeadObjectOutput, *ResourceE
 			return nil, awsErrorToException(err)
 		}
 		r.cachedHead = head
+		if r.heads != nil {
+			r.heads.Store(r.key, head)
+		}
 	}
 	return r.cachedHead, nil
 }

pkg/parser/audio/mp4.go

@@ -277,6 +277,14 @@ func readChapterTitles(ctx context.Context, res fetcher.Resource, offsets []uint
 			if cache != nil {
 				if b, hit := cache.cachedSlice(run.lo, run.hi); hit {
 					data = b
+				} else if cache.retain {
+					// The cache is being retained for serving the publication:
+					// pull whole blocks through it (costing up to a block of
+					// extra transfer per run) so the ranges a browser requests
+					// around each chapter are later served from memory.
+					if b, err := cache.Read(ctx, run.lo, run.hi); err == nil {
+						data = b
+					}
 				}
 			}
 			if data == nil {

pkg/parser/audio/parser.go

@@ -38,6 +38,11 @@ type AudioParser struct {
 	// samples). Zero means the default. Higher values hide more per-file latency
 	// on remote sources at the cost of more in-flight requests.
 	concurrency int
+
+	// retainCache keeps the blocks fetched while probing attached to the
+	// publication, so serving the audio files can answer those byte ranges from
+	// memory. See WithRetainedCache.
+	retainCache bool
 }
 
 // Option configures an AudioParser.
@@ -79,6 +84,24 @@ func WithConcurrency(n int) Option {
 	}
 }
 
+// WithRetainedCache keeps the read-cache blocks fetched while probing attached
+// to the parsed publication, so that serving its audio files answers those
+// byte ranges — and the resources' lengths — from memory instead of new remote
+// requests. Chapter-title samples are also pulled through the block cache
+// (costing up to one block of extra transfer per chapter while parsing).
+//
+// The retained ranges — container headers and chapter samples — are exactly
+// what a browser's demuxer requests before starting playback of an M4B, so
+// this trades a slightly more expensive parse for a much faster time to first
+// audio. Recommended when the publication is opened once and served many
+// times (e.g. an HTTP streamer in front of remote storage); wasteful for
+// one-shot parsing, and unnecessary for local files. Memory cost is bounded
+// by what probing touched: typically a few blocks per file plus one per
+// chapter.
+func WithRetainedCache() Option {
+	return func(p *AudioParser) { p.retainCache = true }
+}
+
 func NewParser() AudioParser {
 	return AudioParser{}
 }
@@ -147,9 +170,14 @@ func (p AudioParser) Parse(ctx context.Context, asset asset.PublicationAsset, fe
 	// When rich parsing is enabled, probe the audio files for durations,
 	// bitrates, embedded metadata, a cover and a table of contents.
 	if p.rich {
-		if coverFactory := p.enrich(ctx, fetcher, &man); coverFactory != nil {
+		coverFactory, caches := p.enrich(ctx, fetcher, &man)
+		if coverFactory != nil {
 			serviceFactories[pub.CoverService_Name] = coverFactory
 		}
+		if len(caches) > 0 {
+			// Serve the byte ranges fetched while probing from memory.
+			fetcher = &cacheFetcher{Fetcher: fetcher, caches: caches}
+		}
 	}
 
 	var builder *pub.ServicesBuilder