diff --git a/js/components/src/components/mobile-player/shared.tsx b/js/components/src/components/mobile-player/shared.tsx index 33906e747..e0f84fd60 100644 --- a/js/components/src/components/mobile-player/shared.tsx +++ b/js/components/src/components/mobile-player/shared.tsx @@ -45,7 +45,7 @@ export function srcToUrl( } let outUrl: string; if (protocol === PlayerProtocol.HLS) { - const llhlsUrl = `${url}/api/playback/${encodeURIComponent(props.src)}/llhls/master.m3u8`; + const llhlsUrl = `${url}/api/playback/${encodeURIComponent(props.src)}/llhls/main.m3u8`; if ( props.selectedRendition && props.selectedRendition !== "auto" && diff --git a/pkg/api/playback_llhls.go b/pkg/api/playback_llhls.go index da71eba34..d81171527 100644 --- a/pkg/api/playback_llhls.go +++ b/pkg/api/playback_llhls.go @@ -144,7 +144,7 @@ func (a *StreamplaceAPI) HandleLLHLS(ctx context.Context) httprouter.Handle { parts := strings.Split(path, "/") params := httprouter.Params{{Key: "user", Value: p.ByName("user")}} switch { - case path == "master.m3u8": + case path == "main.m3u8": master(w, r, params) case len(parts) == 3 && parts[2] == "index.m3u8": params = append(params, @@ -334,8 +334,7 @@ func (a *StreamplaceAPI) HandleLLHLSPart(ctx context.Context) httprouter.Handle } func validLLHLSFrameRate(fps float64) bool { - // Apple HLS authoring rules cap advertised video frame rates at 60 fps. - return fps > 0 && fps <= 60 && !math.IsNaN(fps) && !math.IsInf(fps, 0) + return fps > 0 && !math.IsNaN(fps) && !math.IsInf(fps, 0) } func (a *StreamplaceAPI) HandleLLHLSSegment(ctx context.Context) httprouter.Handle { diff --git a/pkg/api/playback_llhls_test.go b/pkg/api/playback_llhls_test.go index 63e075c8d..34ac562ce 100644 --- a/pkg/api/playback_llhls_test.go +++ b/pkg/api/playback_llhls_test.go @@ -135,7 +135,7 @@ func TestLLHLSMasterReturnsUnavailableWithoutMeasuredBandwidth(t *testing.T) { recorder := httptest.NewRecorder() requestContext, cancel := context.WithTimeout(context.Background(), time.Millisecond) defer cancel() - request := httptest.NewRequest(http.MethodGet, "/master.m3u8", nil).WithContext(requestContext) + request := httptest.NewRequest(http.MethodGet, "/main.m3u8", nil).WithContext(requestContext) api.HandleLLHLSMaster(context.Background())(recorder, request, httprouter.Params{{Key: "user", Value: user}}) if recorder.Code != http.StatusServiceUnavailable { t.Fatalf("master without measured bandwidth status = %d, want %d", recorder.Code, http.StatusServiceUnavailable) @@ -143,7 +143,7 @@ func TestLLHLSMasterReturnsUnavailableWithoutMeasuredBandwidth(t *testing.T) { } func TestLLHLSMasterRedirectsForInvalidVideoFrameRate(t *testing.T) { - for _, fps := range []float64{math.NaN(), math.Inf(1), -1, 61} { + for _, fps := range []float64{math.NaN(), math.Inf(1), -1} { t.Run(fmt.Sprintf("fps-%v", fps), func(t *testing.T) { const user = "did:key:z6MkInvalidFrameRateTest" window := llhls.NewWindow() @@ -158,7 +158,7 @@ func TestLLHLSMasterRedirectsForInvalidVideoFrameRate(t *testing.T) { setLLWindowsForTest(manager, map[string]*llhls.Window{user: window}) api := &StreamplaceAPI{MediaManager: manager, Aliases: map[string]string{}} recorder := httptest.NewRecorder() - api.HandleLLHLSMaster(context.Background())(recorder, httptest.NewRequest(http.MethodGet, "/master.m3u8", nil), httprouter.Params{{Key: "user", Value: user}}) + api.HandleLLHLSMaster(context.Background())(recorder, httptest.NewRequest(http.MethodGet, "/main.m3u8", nil), httprouter.Params{{Key: "user", Value: user}}) if recorder.Code != http.StatusTemporaryRedirect { t.Fatalf("master with invalid frame rate status = %d, want %d", recorder.Code, http.StatusTemporaryRedirect) } @@ -171,14 +171,14 @@ func TestLLHLSMasterAdvertisesFrameRateWhenKnown(t *testing.T) { Codec: "avc1.64002a", Width: 1280, Height: 720, - FrameRate: 59.94, + FrameRate: 120, Bandwidth: 5000000, AverageBandwidth: 4000000, }, llhls.AudioConfig{Channels: 2, Bandwidth: 128000, AverageBandwidth: 128000}) if err != nil { t.Fatal(err) } - if !strings.Contains(master, "FRAME-RATE=59.940") { + if !strings.Contains(master, "FRAME-RATE=120.000") { t.Fatalf("master omitted known video frame rate:\n%s", master) } } @@ -224,7 +224,7 @@ func TestLLHLSMasterRedirectsWhilePresentationIsInitializing(t *testing.T) { handler := api.HandleLLHLSMaster(context.Background()) recorder := httptest.NewRecorder() - request := httptest.NewRequest(http.MethodGet, "/master.m3u8", nil) + request := httptest.NewRequest(http.MethodGet, "/main.m3u8", nil) handler(recorder, request, httprouter.Params{{Key: "user", Value: user}}) if recorder.Code != http.StatusTemporaryRedirect { @@ -239,13 +239,41 @@ func TestLLHLSMasterRedirectsWhilePresentationIsInitializing(t *testing.T) { } } +func TestHandleLLHLSUsesMainPlaylistName(t *testing.T) { + const user = "did:key:z6MkMainPlaylistRouteTest" + manager := &media.MediaManager{} + setLLWindowsForTest(manager, map[string]*llhls.Window{user: llhls.NewWindow()}) + api := &StreamplaceAPI{MediaManager: manager, Aliases: map[string]string{}} + handler := api.HandleLLHLS(context.Background()) + + for _, test := range []struct { + name string + path string + want int + }{ + {name: "main", path: "/main.m3u8", want: http.StatusTemporaryRedirect}, + {name: "master", path: "/master.m3u8", want: http.StatusNotFound}, + } { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + handler(recorder, httptest.NewRequest(http.MethodGet, test.path, nil), httprouter.Params{ + {Key: "user", Value: user}, + {Key: "path", Value: test.path}, + }) + if recorder.Code != test.want { + t.Fatalf("%s status = %d, want %d", test.path, recorder.Code, test.want) + } + }) + } +} + func TestLLHLSMasterWaitsForBothRenditions(t *testing.T) { const user = "did:key:z6MkBothTracksTest" window := llhls.NewWindow() if err := window.Observe(llhls.Event{Kind: llhls.Init, Presentation: "p", Track: "video", Generation: 1, Data: []byte("video-init")}); err != nil { t.Fatal(err) } - window.SetVideoConfig(llhls.VideoConfig{FrameRate: 30, Bandwidth: 5000000, AverageBandwidth: 4000000}) + window.SetVideoConfig(llhls.VideoConfig{FrameRate: 120, Bandwidth: 5000000, AverageBandwidth: 4000000}) manager := &media.MediaManager{} setLLWindowsForTest(manager, map[string]*llhls.Window{user: window}) api := &StreamplaceAPI{MediaManager: manager, Aliases: map[string]string{}} @@ -254,7 +282,7 @@ func TestLLHLSMasterWaitsForBothRenditions(t *testing.T) { result := make(chan *httptest.ResponseRecorder, 1) go func() { recorder := httptest.NewRecorder() - handler(recorder, httptest.NewRequest(http.MethodGet, "/master.m3u8", nil), httprouter.Params{{Key: "user", Value: user}}) + handler(recorder, httptest.NewRequest(http.MethodGet, "/main.m3u8", nil), httprouter.Params{{Key: "user", Value: user}}) result <- recorder }() select { @@ -273,7 +301,7 @@ func TestLLHLSMasterWaitsForBothRenditions(t *testing.T) { case <-time.After(time.Second): t.Fatal("master did not become ready after both renditions initialized") } - if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), `AUDIO="audio"`) { + if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), `AUDIO="audio"`) || !strings.Contains(recorder.Body.String(), "FRAME-RATE=120.000") { t.Fatalf("master after both init segments = status %d body %q", recorder.Code, recorder.Body.String()) } } @@ -293,7 +321,7 @@ func TestLLHLSMasterRedirectsWithoutVideoFrameRate(t *testing.T) { recorder := httptest.NewRecorder() requestContext, cancel := context.WithTimeout(context.Background(), time.Millisecond) defer cancel() - request := httptest.NewRequest(http.MethodGet, "/master.m3u8", nil).WithContext(requestContext) + request := httptest.NewRequest(http.MethodGet, "/main.m3u8", nil).WithContext(requestContext) api.HandleLLHLSMaster(context.Background())(recorder, request, httprouter.Params{{Key: "user", Value: user}}) if recorder.Code != http.StatusServiceUnavailable { t.Fatalf("master without frame rate status = %d, want %d", recorder.Code, http.StatusServiceUnavailable) diff --git a/pkg/llhls/window.go b/pkg/llhls/window.go index 287b20423..863bc46f5 100644 --- a/pkg/llhls/window.go +++ b/pkg/llhls/window.go @@ -1,5 +1,3 @@ -// Package llhls contains the bounded in-memory state used by the LL-HLS -// origin. It deliberately knows nothing about GStreamer or HTTP. package llhls import ( @@ -171,7 +169,7 @@ type Option func(*Window) func WithMaxSegments(n int) Option { return func(w *Window) { w.maxSegments = n } } func WithMaxBytes(n int) Option { return func(w *Window) { w.maxBytes = n } } -// WithPartHoldBack sets the advertised live edge distance for LL-HLS parts. +// Sets the advertised live edge distance for LL-HLS parts. // Values below the HLS minimum are raised to three part targets. func WithPartHoldBack(d time.Duration) Option { return func(w *Window) { @@ -181,7 +179,7 @@ func WithPartHoldBack(d time.Duration) Option { } } -// WithSegmentCompletionDelay delays parent completion visibility by d. This +// Delays parent completion visibility by d. This // gives blocking reloads time to observe the final part of an open segment. func WithSegmentCompletionDelay(d time.Duration) Option { return func(w *Window) { w.completionHold = d } @@ -195,7 +193,7 @@ func withMaxWaiters(n int) Option { } } -// WithPlaylistDurations sets fixed upper bounds for parent and part durations. +// Sets fixed upper bounds for parent and part durations. // The parent bound is rounded to the nearest whole second for TARGETDURATION. // Both values remain fixed for each presentation observed by the Window. func WithPlaylistDurations(parent, part time.Duration) Option { @@ -377,7 +375,7 @@ func (w *Window) findOrCreateSegment(t *track, ev Event) *segment { return s } -// scheduleCompletion flips a closing segment to complete after the hold. The +// Flips a closing segment to complete after the hold. The // timer re-validates under the lock: a presentation reset or discontinuity // may have dropped the segment (or the whole track) in the meantime. func (w *Window) scheduleCompletion(t *track, trackID string, s *segment) { @@ -409,7 +407,7 @@ func (w *Window) completeClosingSegments(t *track, nextMSN uint64) { func (w *Window) evict() { for _, t := range w.tracks { for len(t.segments) > 0 { - // Incomplete segments are still receiving part events; evicting + // Incomplete segments are still receiving part events, so evicting // one strands those events and fails the ingest stream. Small // tracks would otherwise be emptied wholesale while a large // track keeps the byte budget exceeded. @@ -650,10 +648,8 @@ func (w *Window) SegmentData(presentation, trackID string, msn uint64) []byte { return nil } -// Wait blocks a playlist reload until the requested reload point is reflected -// in a newer playlist. For a completed parent, a part index beyond the final -// part rolls over to part zero of the following parent, per HLS blocking -// reload semantics. This rule intentionally does not apply to media lookups. +// Blocks a playlist reload until the requested reload point is reflected +// in a newer playlist, except for media lookups. func (w *Window) Wait(ctx context.Context, presentation, trackID string, msn uint64, partIndex uint32) error { return w.waitForChange(ctx, w.waiters, func() (bool, error) { if w.reloadPointUnavailableLocked(presentation, trackID, msn) || w.reloadPartUnavailableLocked(presentation, trackID, msn, partIndex) { @@ -663,7 +659,7 @@ func (w *Window) Wait(ctx context.Context, presentation, trackID string, msn uin }) } -// WaitForMaster blocks until both renditions and the metadata required by the +// Blocks until both renditions and the metadata required by the // multivariant playlist have been published for a presentation. func (w *Window) WaitForMaster(ctx context.Context, presentation string) error { return w.waitForChange(ctx, w.masterWaiters, func() (bool, error) { @@ -677,9 +673,8 @@ func (w *Window) WaitForMaster(ctx context.Context, presentation string) error { func (w *Window) masterMetadataReadyLocked() bool { video := w.tracks["video"] audio := w.tracks["audio"] - // Keep the master within Apple's HLS authoring limit for advertised frame rate. return video != nil && len(video.init) > 0 && audio != nil && len(audio.init) > 0 && - w.videoConfig.FrameRate > 0 && w.videoConfig.FrameRate <= 60 && !math.IsNaN(w.videoConfig.FrameRate) && !math.IsInf(w.videoConfig.FrameRate, 0) && + w.videoConfig.FrameRate > 0 && !math.IsNaN(w.videoConfig.FrameRate) && !math.IsInf(w.videoConfig.FrameRate, 0) && w.videoConfig.Bandwidth > 0 && w.videoConfig.AverageBandwidth > 0 && w.audioConfig.Bandwidth > 0 && w.audioConfig.AverageBandwidth > 0 } @@ -873,10 +868,8 @@ func acquireWaiter(waiters chan struct{}) error { func releaseWaiter(waiters chan struct{}) { <-waiters } -// Playlist renders the media playlist for one track. URIs are supplied by the +// Renders the media playlist for one track. URIs are supplied by the // caller so routing and presentation identifiers remain outside this package. -// When renditionURI is non-nil, an EXT-X-RENDITION-REPORT is emitted for -// every other track that has published media (required for LL-HLS). func (w *Window) Playlist(presentation, trackID string, partURI func(uint64, uint32) string, segmentURI func(uint64) string, initURI string, renditionURI func(string) string) string { s := w.playlistSnapshot(presentation, trackID) if s.Track == "" { @@ -890,6 +883,8 @@ func (w *Window) Playlist(presentation, trackID string, partURI func(uint64, uin b.WriteString("#EXT-X-INDEPENDENT-SEGMENTS\n") } fmt.Fprintf(&b, "#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=%.6f,HOLD-BACK=%.6f\n", partHoldBack.Seconds(), 3*float64(targetSeconds)) + // When renditionURI is non-nil, an EXT-X-RENDITION-REPORT is emitted for + // every other track that has published media (required for LL-HLS). if renditionURI != nil { for _, rep := range w.renditionReports(presentation, trackID) { fmt.Fprintf(&b, "#EXT-X-RENDITION-REPORT:URI=%q,LAST-MSN=%d", renditionURI(rep.trackID), rep.lastMSN) @@ -971,10 +966,8 @@ type renditionReport struct { lastPart int32 } -// renditionReports describes the latest published state of every track other -// than exclude. A track with an open segment reports that segment's MSN and -// its last published part; a fully completed tail reports its last MSN -// without a part (completed parents carry no listed partial segments). +// Describes the latest published state of every track other +// than exclude func (w *Window) renditionReports(presentation, exclude string) []renditionReport { w.mu.Lock() defer w.mu.Unlock() diff --git a/pkg/media/cmaf_irregular_gop_test.go b/pkg/media/cmaf_irregular_gop_test.go new file mode 100644 index 000000000..1ac7b6f49 --- /dev/null +++ b/pkg/media/cmaf_irregular_gop_test.go @@ -0,0 +1,172 @@ +package media + +import ( + "bytes" + "testing" + "time" + + "github.com/go-gst/go-gst/gst" + "github.com/go-gst/go-gst/gst/app" + "stream.place/streamplace/pkg/gstinit" + "stream.place/streamplace/pkg/llhls" +) + +func TestISOFMP4MuxIrregularGOPWithAudio(t *testing.T) { + gstinit.InitGST() + if gst.Find("isofmp4mux") == nil || gst.Find("x264enc") == nil || gst.Find("fdkaacenc") == nil { + t.Skip("static GStreamer build with isofmp4mux, x264enc, and fdkaacenc is required") + } + + // A 47-frame GOP is about 1.57 seconds at 30 fps, so it divides neither the + // 2-second parent nor the 1-second part. Disabling force-keyunit preserves + // that irregular cadence and makes the muxer handle GOPs crossing fragment + // boundaries. + pipeline, err := gst.NewPipelineFromString( + "isofmp4mux name=mux fragment-duration=2000000000 chunk-duration=1000000000 send-force-keyunit=false ! appsink name=sink sync=false\n" + "videotestsrc num-buffers=300 is-live=true pattern=ball ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc tune=zerolatency speed-preset=ultrafast bframes=0 key-int-max=47 ! h264parse ! video/x-h264,stream-format=avc,alignment=au ! queue ! mux.\n" + "audiotestsrc num-buffers=480 is-live=true samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc bitrate=128000 ! aacparse ! audio/mpeg,mpegversion=4,stream-format=raw,rate=48000,channels=2 ! queue ! mux.", + ) + if err != nil { + t.Fatal(err) + } + sinkElement, err := pipeline.GetElementByName("sink") + if err != nil { + t.Fatal(err) + } + + sink := app.SinkFromElement(sinkElement) + sink.SetBufferListSupport(true) + state := &cmafTrackSink{ + presentation: "test", + track: "av", + window: llhls.NewWindow(), + partDuration: time.Second, + } + callbackErr := make(chan error, 1) + done := make(chan struct{}) + sink.SetCallbacks(&app.SinkCallbacks{ + NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { + sample := sink.PullSample() + if sample == nil { + return gst.FlowEOS + } + if err := state.sample(sample); err != nil { + select { + case callbackErr <- err: + default: + } + return gst.FlowError + } + return gst.FlowOK + }, + EOSFunc: func(*app.Sink) { close(done) }, + }) + + if err := pipeline.SetState(gst.StatePlaying); err != nil { + t.Fatal(err) + } + select { + case <-done: + case <-time.After(15 * time.Second): + _ = pipeline.SetState(gst.StateNull) + t.Fatal("timed out waiting for irregular-GOP isofmp4mux EOS") + } + _ = pipeline.SetState(gst.StateNull) + select { + case err := <-callbackErr: + t.Fatal(err) + default: + } + + snapshot := state.window.Snapshot("test", "av") + if len(snapshot.Init) == 0 || !bytes.Contains(snapshot.Init, []byte("vide")) || !bytes.Contains(snapshot.Init, []byte("soun")) { + t.Fatalf("muxed init does not describe both tracks: %q", snapshot.Init) + } + if len(snapshot.Segments) < 2 { + t.Fatalf("expected at least two completed parent segments, got %d", len(snapshot.Segments)) + } + + var previousEnd time.Duration + havePart := false + previousTiming := make(map[uint32]cmafFragmentTiming) + const maxSingleTrackTail = 50 * time.Millisecond + singleTrackTailDuration := time.Duration(0) + completedSegments := 0 + for _, segment := range snapshot.Segments { + if segment.Complete { + completedSegments++ + if segment.Duration <= 0 || len(segment.Data) == 0 { + t.Fatalf("parent segment %d has no completed CMAF data: duration=%s data=%d", segment.MSN, segment.Duration, len(segment.Data)) + } + if len(segment.Parts) < 2 { + t.Fatalf("parent segment %d has %d parts, want multiple parts", segment.MSN, len(segment.Parts)) + } + parentTimings, err := inspectCMAFFragment(segment.Data) + if err != nil { + t.Fatalf("parent segment %d is not parseable CMAF: %v", segment.MSN, err) + } + if len(parentTimings) < 2 { + t.Fatalf("parent segment %d does not contain both track fragments", segment.MSN) + } + } + + for _, part := range segment.Parts { + if part.Duration <= 0 || len(part.Data) == 0 { + t.Fatalf("parent segment %d part %d is empty or has invalid duration %s", segment.MSN, part.Index, part.Duration) + } + if part.Duration > 1100*time.Millisecond { + t.Fatalf("parent segment %d part %d exceeds the configured 1s chunk duration: %s", segment.MSN, part.Index, part.Duration) + } + partTimings, err := inspectCMAFFragment(part.Data) + if err != nil { + t.Fatalf("parent segment %d part %d is not parseable CMAF: %v", segment.MSN, part.Index, err) + } + switch len(partTimings) { + case 1: + singleTrackTailDuration += part.Duration + if part.Duration > maxSingleTrackTail || singleTrackTailDuration > maxSingleTrackTail { + t.Fatalf("parent segment %d part %d has a meaningful single-track tail: duration=%s consecutive=%s", segment.MSN, part.Index, part.Duration, singleTrackTailDuration) + } + case 2: + singleTrackTailDuration = 0 + default: + t.Fatalf("parent segment %d part %d has %d track fragments, want audio and video", segment.MSN, part.Index, len(partTimings)) + } + if havePart && part.Start != previousEnd { + t.Fatalf("part timeline is not contiguous: parent=%d part=%d got start=%s, want=%s", segment.MSN, part.Index, part.Start, previousEnd) + } + previousEnd = part.Start + part.Duration + havePart = true + + boxTypes := make(map[string]bool) + if err := walkCMAFBoxes(part.Data, func(boxType string, _ []byte) error { + boxTypes[boxType] = true + return nil + }); err != nil { + t.Fatalf("parent segment %d part %d has invalid CMAF boxes: %v", segment.MSN, part.Index, err) + } + if !boxTypes["moof"] || !boxTypes["mdat"] { + t.Fatalf("parent segment %d part %d is missing moof or mdat: %v", segment.MSN, part.Index, boxTypes) + } + + seenTracks := make(map[uint32]bool) + for _, timing := range partTimings { + if seenTracks[timing.TrackID] { + t.Fatalf("parent segment %d part %d repeats track id %d", segment.MSN, part.Index, timing.TrackID) + } + seenTracks[timing.TrackID] = true + if previous, ok := previousTiming[timing.TrackID]; ok { + expected := previous.DecodeTime + previous.Duration + if timing.DecodeTime != expected { + t.Fatalf("track %d decode timeline is not contiguous at parent %d part %d: got %d, want %d", timing.TrackID, segment.MSN, part.Index, timing.DecodeTime, expected) + } + } + previousTiming[timing.TrackID] = timing + } + } + } + if completedSegments < 2 { + t.Fatalf("expected at least two completed parent segments, got %d", completedSegments) + } + if len(previousTiming) != 2 { + t.Fatalf("expected exactly two contiguous decode timelines, got %d: %v", len(previousTiming), previousTiming) + } +}