diff --git a/pkg/media/audio_stems_test.go b/pkg/media/audio_stems_test.go index c5d4b3c4..5d72014f 100644 --- a/pkg/media/audio_stems_test.go +++ b/pkg/media/audio_stems_test.go @@ -316,9 +316,17 @@ func TestWebRTCPlayback2StemTracks(t *testing.T) { // names in the fMP4 reach the signed catalogs. func groupedSignedSegment(t *testing.T, ctx context.Context) []byte { t.Helper() - labeled, err := muxl.RunMuxlCanonicalize(ctx, groupedTestFMP4(t, ctx), nil, muxl.WithTrackMeta(groupedTestMeta)) + first := signLabeledSegments(t, ctx, newBareSegmentSigner(t), groupedTestFMP4(t, ctx), groupedTestMeta)[0] + requireGroupedCatalog(t, ctx, first) + return first +} + +// signLabeledSegments writes meta into frag's tracks and signs every GoP the +// way fmp4-direct ingest does, returning the bare signed segments. +func signLabeledSegments(t *testing.T, ctx context.Context, ms *MediaSignerLocal, frag []byte, meta map[uint32]muxl.TrackMeta) [][]byte { + t.Helper() + labeled, err := muxl.RunMuxlCanonicalize(ctx, frag, nil, muxl.WithTrackMeta(meta)) require.NoError(t, err) - ms := newBareSegmentSigner(t) eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { @@ -326,15 +334,62 @@ func groupedSignedSegment(t *testing.T, ctx context.Context) []byte { close(eventCh) errCh <- err }() - var first []byte + var segs [][]byte for ev := range eventCh { - if ev.Type == "signed-segment" && first == nil { - first = concatTracksByID(ev.Tracks) + if ev.Type == "signed-segment" { + segs = append(segs, concatTracksByID(ev.Tracks)) } } require.NoError(t, <-errCh) - requireGroupedCatalog(t, ctx, first) - return first + require.NotEmpty(t, segs) + return segs +} + +// Codec completion encodes the main group's track even when a stem comes +// first in track-id order: a mono stem on track 2 ahead of the stereo, +// AAC-only full mix on track 3. The completed Opus must be the mix's. +func TestStreamTranscoderCompletesTheMainGroup(t *testing.T) { + ctx := context.Background() + frag := runSynthPipeline(t, ctx, strings.Join([]string{ + "videotestsrc num-buffers=60 ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc key-int-max=30 tune=zerolatency speed-preset=ultrafast ! h264parse ! mp4mux name=mux fragment-duration=500 ! appsink name=sink", + "audiotestsrc num-buffers=94 samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=1 ! audioconvert ! fdkaacenc ! aacparse ! mux.", + "audiotestsrc num-buffers=94 samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", + }, "\n")) + ms := newBareSegmentSigner(t) + segs := signLabeledSegments(t, ctx, ms, frag, map[uint32]muxl.TrackMeta{ + 2: {AlternateGroup: 2, Label: "Music"}, + 3: {AlternateGroup: 1, Label: "Mix"}, + }) + + keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) + require.NoError(t, err) + mm := &MediaManager{cli: &config.CLI{BroadcasterHost: "test.example.com"}} + // Close waits for the pipeline, which delivers every completion first. + var completed [][]byte + tr := mm.newStreamTranscoder(ctx, "opus", ms.Cert, keyPEM, func(_ any, c []byte) { + completed = append(completed, c) + }) + for i, s := range segs { + require.NoError(t, tr.Feed(s, i)) + } + require.NoError(t, tr.Close()) + require.NotEmpty(t, completed) + + for i, c := range completed { + events, err := unwrapMuxlEvents(ctx, c) + require.NoError(t, err) + cat, _ := catalogAndTracks(events) + var opus []muxl.MuxlAudioConfig + for _, a := range audioTracks(cat) { + if isOpusCodec(a.Codec) { + opus = append(opus, a) + } + } + require.Len(t, opus, 1, "segment %d", i) + require.Equal(t, uint32(2), opus[0].NumberOfChannels, "segment %d: Opus encoded from the stereo mix, not the mono stem", i) + require.Equal(t, uint32(1), opus[0].AlternateGroup, "segment %d", i) + require.Equal(t, "Mix", opus[0].Label, "segment %d", i) + } } // A syndicating node pairs a rendition with the source segment's Opus diff --git a/pkg/media/transcode_stream.go b/pkg/media/transcode_stream.go index 4585aac3..27e400eb 100644 --- a/pkg/media/transcode_stream.go +++ b/pkg/media/transcode_stream.go @@ -249,11 +249,20 @@ func (t *streamTranscoder) Feed(src []byte, token any) error { t.started = true t.mu.Unlock() + // The pipeline re-encodes the demux's first audio stream, so hand it only + // video and the main group's source track: a stem may precede it in + // track-id order. The job keeps the whole source to append the result to. + wantOpus := t.target == "aac" + input, err := filterSegmentToCodec(t.ctx, src, wantOpus) + if err != nil { + return fmt.Errorf("select transcoder source audio: %w", err) + } + // Synthesize and write the init (ftyp+moov) once, then the segment bytes — // the same init-then-blind-concat the RTMP push feeder uses. if first { var init bytes.Buffer - if err := muxl.RunMuxlWrapInit(t.ctx, bytes.NewReader(src), &init); err != nil { + if err := muxl.RunMuxlWrapInit(t.ctx, bytes.NewReader(input), &init); err != nil { return fmt.Errorf("synthesize transcoder init: %w", err) } if _, err := t.feedW.Write(init.Bytes()); err != nil { @@ -267,7 +276,7 @@ func (t *streamTranscoder) Feed(src []byte, token any) error { case <-t.ctx.Done(): return t.ctx.Err() } - if _, err := t.feedW.Write(src); err != nil { + if _, err := t.feedW.Write(input); err != nil { return err } return nil @@ -440,10 +449,10 @@ func (mm *MediaManager) finishTranscodedSegment(ctx context.Context, srcSeg, tra if cat == nil || cat.Audio == nil { return nil, fmt.Errorf("source segment has no audio track") } - // The transcoder re-encodes the demux's first audio stream, the main - // group's lowest track (completion runs only when the main group lacks a - // codec, see completionTarget); the completed track joins that group so - // the main-group consumers find it beside the source track. + // The transcoder re-encodes the main group's source track (Feed filters + // its input to it; completion runs only when the main group lacks a codec, + // see completionTarget); the completed track joins that group so the + // main-group consumers find it beside the source track. main := mainAudioTracks(audioTracks(cat)) if len(main) == 0 { return nil, fmt.Errorf("source segment has no audio track")