diff --git a/go.mod b/go.mod index f775d595..2efdce6d 100644 --- a/go.mod +++ b/go.mod @@ -71,7 +71,7 @@ require ( github.com/streamplace/atmoq/go v0.0.4-0.20260701223355-13757de4ae08 github.com/streamplace/atproto-oauth-golang v0.0.0-20260413212710-98956064d06c github.com/streamplace/glex v0.0.0-20260716203108-f73ed7cc31c9 - github.com/streamplace/muxl/go v0.3.4 + github.com/streamplace/muxl/go v0.3.5-0.20260726020638-c7539b84f0b1 github.com/streamplace/oatproxy v0.0.0-20260710202406-60d97b9d780b github.com/stretchr/testify v1.11.1 github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71 diff --git a/go.sum b/go.sum index 5f890882..2b4b0fd9 100644 --- a/go.sum +++ b/go.sum @@ -1383,6 +1383,8 @@ github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507 h1:e8M3qPLr37Nx github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507/go.mod h1:Pm2I1+iDXn/hLbF7XCg/DsZi6uDCiOo7hZGWprSM7k0= github.com/streamplace/muxl/go v0.3.4 h1:M/G8CRKjAmfsyWtolLziUtNrNtS3lJFq/H6IhnY1yBo= github.com/streamplace/muxl/go v0.3.4/go.mod h1:aCyYTW3o6c1Kush9UJ/Yv6EYMUbj8l8GTD7cHKcSxw8= +github.com/streamplace/muxl/go v0.3.5-0.20260726020638-c7539b84f0b1 h1:Jqr7+PhzDQ6pz41vAvelmLuoen2mevs9KB6xUG2ogGE= +github.com/streamplace/muxl/go v0.3.5-0.20260726020638-c7539b84f0b1/go.mod h1:aCyYTW3o6c1Kush9UJ/Yv6EYMUbj8l8GTD7cHKcSxw8= github.com/streamplace/oatproxy v0.0.0-20260710202406-60d97b9d780b h1:eWbwCtBbMyrDTHLYIold07OR2hmvzXsbAUxi57ElMLk= github.com/streamplace/oatproxy v0.0.0-20260710202406-60d97b9d780b/go.mod h1:wpY+T/wE00jrUhgh2dKXbbE91D36u86KGlENK/hWFkE= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/pkg/muxl/muxl.go b/pkg/muxl/muxl.go index 30ec98fa..11b2166a 100644 --- a/pkg/muxl/muxl.go +++ b/pkg/muxl/muxl.go @@ -158,6 +158,42 @@ func RunMuxlUnwrapEvents(ctx context.Context, input io.Reader, eventCh chan *Mux return eng.UnwrapEvents(ctx, input, eventCh) } +// RunMuxlReadSegments reads count canonical segments out of a stored MUXL +// wrapper, starting at offset bytes into its canonical-segment stream, and +// returns them verbatim. +// +// offset is fragment-relative — the offset a Metafile segment records — NOT an +// absolute offset into the blob. muxl resolves the container framing itself +// (for the flat-MP4 VOD shape, [flat-header][fragments], that means skipping +// the synthesized header), so callers index fragments and never add a header +// size of their own. An offset that misses a segment boundary is an error +// rather than a read of whatever bytes happen to be there. +// +// No length is passed: muxl derives each segment's extent from its own uuid +// boundaries. src is read through a random-access handle, so only the +// requested segments' bytes are fetched — one GoP out of a multi-gigabyte VOD +// costs a few small reads (range GETs against an S3-backed blob). +func RunMuxlReadSegments(ctx context.Context, src io.ReaderAt, size, offset int64, count int) ([]byte, error) { + eng, err := getEngine() + if err != nil { + return nil, err + } + return eng.ReadSegments(ctx, src, size, offset, count) +} + +// RunMuxlReadSegmentsAtFileOffset is RunMuxlReadSegments for an index whose +// offsets are absolute positions in the blob rather than fragment-relative — +// the legacy [init][segments] shape, whose Metafile offsets already count the +// leading init (see Metafile.FlatHeaderSize). muxl adds nothing to the offset +// but still derives segment extents and rejects a bad one. +func RunMuxlReadSegmentsAtFileOffset(ctx context.Context, src io.ReaderAt, size, offset int64, count int) ([]byte, error) { + eng, err := getEngine() + if err != nil { + return nil, err + } + return eng.ReadSegments(ctx, src, size, offset, count, upstream.WithFileOffset()) +} + // RunMuxlSegmenterEvents segments an fMP4 stream into per-GoP canonical MUXL // events (unsigned). func RunMuxlSegmenterEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent) error { diff --git a/pkg/vod/thumbnail.go b/pkg/vod/thumbnail.go index b99154f9..0ba5ff8a 100644 --- a/pkg/vod/thumbnail.go +++ b/pkg/vod/thumbnail.go @@ -13,6 +13,7 @@ import ( "stream.place/streamplace/pkg/blob" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" + "stream.place/streamplace/pkg/muxl" ) // thumbnailFormat / thumbnailMimeType are the encoding used for the @@ -76,8 +77,25 @@ func generateThumbnail(ctx context.Context, store blob.Store, cid string, meta * return nil, fmt.Errorf("open content blob: %w", err) } defer content.Close() - segBytes := make([]byte, seg.Size) - if _, err := io.ReadFull(io.NewSectionReader(content, seg.Offset, seg.Size), segBytes); err != nil { + // Hand muxl the random-access handle and the segment's index coordinates + // and let it produce the bytes, rather than slicing the blob here. Doing + // it by hand meant adding Metafile.FlatHeaderSize to reach the fragments + // past a flat blob's synthesized header, and omitting it read a header's + // worth of the wrong bytes — which a decoder reports only as "this file is + // invalid and cannot be played". muxl owns the layout it synthesized, so + // it owns the arithmetic, derives the segment's extent from its own + // boundaries, and errors loudly on an offset that misses one. + // + // Which coordinate space the offsets are in is the metafile's to declare: + // FlatHeaderSize is set for the [flat-header][fragments] shape, whose + // offsets are fragment-relative, and zero for legacy [init][segments] + // blobs, whose offsets are already absolute. + readSegments := muxl.RunMuxlReadSegments + if meta.FlatHeaderSize == 0 { + readSegments = muxl.RunMuxlReadSegmentsAtFileOffset + } + segBytes, err := readSegments(ctx, content, content.Size(), seg.Offset, 1) + if err != nil { return nil, fmt.Errorf("read segment bytes: %w", err) } // Timing breadcrumb: blob read vs. render (flatten + decode, logged diff --git a/pkg/vod/thumbnail_flat_test.go b/pkg/vod/thumbnail_flat_test.go new file mode 100644 index 00000000..d2847cee --- /dev/null +++ b/pkg/vod/thumbnail_flat_test.go @@ -0,0 +1,128 @@ +package vod + +import ( + "bytes" + "context" + "io" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/blob" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/muxl" +) + +// TestGenerateThumbnailFlatBlob drives the production thumbnail step against a +// real finalized VOD content blob — the [flat-header][fragments] shape, where +// the fragments do not start at byte 0. +// +// That gap is what broke thumbnailing: segment offsets are fragment-relative, +// so reading them as absolute blob offsets lands a whole flat header early and +// the decoder reports only "This file is invalid and cannot be played." +// generateThumbnail now hands the offset to muxl instead of adding the header +// size itself, so this passes without the metafile carrying FlatHeaderSize at +// all — which is exactly the coupling that regressed. +// +// Point SP_THUMBNAIL_BLOB at a blobs/.mp4-shaped file to run it. +func TestGenerateThumbnailFlatBlob(t *testing.T) { + path := os.Getenv("SP_THUMBNAIL_BLOB") + if path == "" { + t.Skip("set SP_THUMBNAIL_BLOB to a finalized VOD content blob") + } + warmGST() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ctx = log.WithLogValues(ctx, "test", "TestGenerateThumbnailFlatBlob") + + abs, err := filepath.Abs(path) + require.NoError(t, err) + info, err := os.Stat(abs) + require.NoError(t, err) + + // The blob's filename is its CID; the store is a temp dir with the blob + // symlinked in so we don't copy gigabytes around. + cid := filepath.Base(abs) + for _, ext := range []string{".m4s", ".mp4"} { + cid = strings.TrimSuffix(cid, ext) + } + + root := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(root, "blobs"), 0755)) + require.NoError(t, os.Symlink(abs, filepath.Join(root, BlobsPrefix+cid+".mp4"))) + + store, err := blob.NewFileStore(root) + require.NoError(t, err) + + // Rebuild the sidecars (metafile + per-track inits) from the blob itself. + // newFragmentMetafileBuilder is what the paths that produce a flat blob + // use (ProcessVOD, finalizeLivestream), so the offsets here are + // fragment-relative exactly as a production metafile's are. + // + // Deliberately NOT regenerateSidecars: that uses newMetafileBuilder, whose + // legacy [init][segments] assumption shifts every offset by the init + // length — a separate transfer-path bug (see transfer.go's follow-up note). + meta, err := buildFragmentMetafile(ctx, t, store, cid, info.Size()) + require.NoError(t, err) + var video *MetafileTrack + for tid := range meta.Tracks { + tr := meta.Tracks[tid] + if tr.Type == "video" { + video = &tr + break + } + } + require.NotNil(t, video, "blob must have a video track") + t.Logf("%d tracks, %d video segments, flatHeaderSize=%d", + len(meta.Tracks), len(video.Segments), meta.FlatHeaderSize) + + thumb, err := generateThumbnail(ctx, store, cid, meta) + require.NoError(t, err) + require.NotEmpty(t, thumb) + require.True(t, bytes.HasPrefix(thumb, []byte{0xFF, 0xD8, 0xFF}), "expected JPEG SOI marker") + t.Logf("thumbnail: %d bytes", len(thumb)) +} + +// buildFragmentMetafile derives a fragment-relative metafile (and the +// per-track init blobs) from a stored content blob, mirroring what the +// flat-blob producers do. +func buildFragmentMetafile(ctx context.Context, t *testing.T, store blob.Store, cid string, size int64) (*Metafile, error) { + t.Helper() + r, err := store.Open(ctx, BlobsPrefix+cid+".mp4") + if err != nil { + return nil, err + } + defer r.Close() + + mb := newFragmentMetafileBuilder(ctx, store) + eventCh := make(chan *muxl.MuxlEvent, 16) + producerErr := make(chan error, 1) + go func() { + producerErr <- muxl.RunMuxlUnwrapEvents(ctx, io.NewSectionReader(r, 0, size), eventCh) + close(eventCh) + }() + for ev := range eventCh { + if e := mb.Observe(ev); e != nil { + return nil, e + } + } + if err := <-producerErr; err != nil { + return nil, err + } + meta := mb.Finalize(cid, size) + + // ProcessVOD/finalizeLivestream set this from the header they synthesized; + // here the blob is already assembled, so recover it as the bytes the + // fragments don't account for ([flat-header][fragments]). + var fragments int64 + for _, tr := range meta.Tracks { + for _, s := range tr.Segments { + fragments += s.Size + } + } + meta.FlatHeaderSize = size - fragments + return meta, nil +}