From 64c79e3dac9babff7edc90b84c8acfbb5777928f Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Tue, 13 Oct 2026 19:22:21 -0700 Subject: [PATCH] Pair the transcoding node's own renditions with the source's stems The transcoder's output carries only the full mix, so a WebRTC viewer who picked a quality on the node that transcoded it got no stems, and silence after mixing one in. Packetize those renditions with the source's Opus tracks, the way replicated renditions already are; the pairing is now shared by both. --- pkg/director/stream_session.go | 8 +++ pkg/media/audio_stems_test.go | 9 +++- pkg/media/rendition_playback.go | 96 ++++++++++++++++++++++----------- 3 files changed, 80 insertions(+), 33 deletions(-) diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index 91ce6388..07deb0a8 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -937,6 +937,10 @@ func (ss *StreamSession) Transcode(ctx context.Context, spseg *placestream.Segme addendum = nil } } + // The transcoder's output carries only the full mix; renditions play + // with the source's own Opus tracks, stems included, wherever the + // addendum lets them be paired. + paired := ss.mm.PacketizeRenditions(ctx, notif.Muxl, addendum) playback := make([]*bus.Seg, len(segs)) var pg errgroup.Group for i, seg := range segs { @@ -958,6 +962,10 @@ func (ss *StreamSession) Transcode(ctx context.Context, spseg *placestream.Segme // with the flag unset was refused to every viewer. Published: notif.Metadata.Published, } + if packet := paired[rs[i].Name]; packet != nil { + playback[i].PacketizedData = packet + continue + } pg.Go(func() error { packet, err := media.Packetize(ctx, ss.cli, playback[i]) if err != nil { diff --git a/pkg/media/audio_stems_test.go b/pkg/media/audio_stems_test.go index 5d72014f..28aea17c 100644 --- a/pkg/media/audio_stems_test.go +++ b/pkg/media/audio_stems_test.go @@ -394,7 +394,8 @@ func TestStreamTranscoderCompletesTheMainGroup(t *testing.T) { // A syndicating node pairs a rendition with the source segment's Opus // tracks: the mix and the stems, so a viewer who picked a rendition keeps -// the stems. +// the stems. So does the node that transcoded it, whose transcoder output +// carries only the mix. func TestPublishRenditionsForPlaybackCarriesStems(t *testing.T) { ctx := context.Background() src := groupedSignedSegment(t, ctx) @@ -424,4 +425,10 @@ func TestPublishRenditionsForPlaybackCarriesStems(t *testing.T) { case <-time.After(20 * time.Second): t.Fatal("no rendition segment published") } + + local := mm.PacketizeRenditions(ctx, src, addendum)["120p"] + require.NotNil(t, local, "the transcoding node pairs its own rendition") + require.NotEmpty(t, local.Video) + require.Len(t, local.Stems, 1) + require.Equal(t, uint32(2), local.Stems[0].Group) } diff --git a/pkg/media/rendition_playback.go b/pkg/media/rendition_playback.go index 1a89ec13..d740535f 100644 --- a/pkg/media/rendition_playback.go +++ b/pkg/media/rendition_playback.go @@ -19,19 +19,18 @@ type sourceAudio struct { opus map[string][]byte // the main group's Opus track and each stem's, canonical bytes by track id } -// rememberSourceAudio keeps a replicated source segment's Opus tracks — the -// main group's and each stem's, so a rendition viewer gets the stems too — -// keyed by its video track's first decode time, for -// PublishRenditionsForPlayback. -func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg []byte) { +// sourceOpus returns a source segment's Opus tracks, the main group's and +// each stem's, so a rendition viewer gets the stems too, with its video +// track's first decode time, which the renditions transcoded from it share. +func sourceOpus(ctx context.Context, seg []byte) (uint64, map[string][]byte, bool) { events, err := unwrapMuxlEvents(ctx, seg) if err != nil { - log.Debug(ctx, "rendition playback: could not read source segment", "streamer", did, "error", err) - return + log.Debug(ctx, "rendition playback: could not read source segment", "error", err) + return 0, nil, false } cat, tracks := catalogAndTracks(events) if cat == nil || cat.Video == nil || cat.Audio == nil || tracks == nil { - return + return 0, nil, false } var videoID string for _, v := range cat.Video.Renditions { @@ -47,7 +46,7 @@ func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg } } if videoID == "" || len(opusIDs) == 0 { - return + return 0, nil, false } for _, a := range stemOpusTracks(all) { opusIDs = append(opusIDs, a.TrackID()) @@ -60,9 +59,17 @@ func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg } } if len(opus) == 0 { - return + return 0, nil, false } tfdt, ok := firstTfdt(tracks[videoID]) + return tfdt, opus, ok +} + +// rememberSourceAudio keeps a replicated source segment's Opus tracks (see +// sourceOpus), keyed by its video's first decode time, for +// PublishRenditionsForPlayback. +func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg []byte) { + tfdt, opus, ok := sourceOpus(ctx, seg) if !ok { return } @@ -115,43 +122,67 @@ func (mm *MediaManager) sourceAudioAt(did string, tfdt uint64) map[string][]byte // node that only syndicates the stream received the renditions for HLS but // had nothing on its bus, so a WebRTC viewer who picked one got no video. func (mm *MediaManager) PublishRenditionsForPlayback(ctx context.Context, did string, addendum []byte, published bool) { - if mm.bus == nil || len(addendum) == 0 { + if mm.bus == nil { return } + packets := mm.renditionPackets(ctx, addendum, func(tfdt uint64) map[string][]byte { + return mm.sourceAudioAt(did, tfdt) + }) + names := make([]string, 0, len(packets)) + for name := range packets { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + mm.bus.PublishSegment(ctx, did, name, &bus.Seg{PacketizedData: packets[name], Published: published}) + } +} + +// PacketizeRenditions packetizes, by rendition name, each rendition of an +// addendum this node minted from source, paired with source's own Opus +// tracks: the mix and the stems, which the transcoder's output lacks. A +// rendition that can't be paired is missing from the result. +func (mm *MediaManager) PacketizeRenditions(ctx context.Context, source, addendum []byte) map[string]*bus.PacketizedSegment { + _, opus, ok := sourceOpus(ctx, source) + if !ok { + return nil + } + return mm.renditionPackets(ctx, addendum, func(uint64) map[string][]byte { return opus }) +} + +// renditionPackets packetizes each rendition track in an addendum together +// with the source Opus tracks opusAt returns for its first decode time, by +// rendition name. +func (mm *MediaManager) renditionPackets(ctx context.Context, addendum []byte, opusAt func(tfdt uint64) map[string][]byte) map[string]*bus.PacketizedSegment { + if len(addendum) == 0 { + return nil + } events, err := unwrapMuxlEvents(ctx, addendum) if err != nil { - log.Warn(ctx, "rendition playback: could not read addendum", "streamer", did, "error", err) - return + log.Warn(ctx, "rendition playback: could not read addendum", "error", err) + return nil } cat, tracks := catalogAndTracks(events) if cat == nil || cat.Video == nil || tracks == nil { - return + return nil } - names := map[string]string{} + packets := map[string]*bus.PacketizedSegment{} for _, v := range cat.Video.Renditions { - if v.CodedHeight > 0 { - names[strconv.FormatUint(uint64(v.TrackID()), 10)] = RenditionName(v.CodedWidth, v.CodedHeight) - } - } - ids := make([]string, 0, len(tracks)) - for id := range tracks { - if names[id] != "" { - ids = append(ids, id) - } - } - sort.Strings(ids) - for _, id := range ids { + id := strconv.FormatUint(uint64(v.TrackID()), 10) video := tracks[id] + if v.CodedHeight == 0 || len(video) == 0 { + continue + } tfdt, ok := firstTfdt(video) if !ok { continue } - opus := mm.sourceAudioAt(did, tfdt) + opus := opusAt(tfdt) if opus == nil { - log.Debug(ctx, "rendition playback: no source audio for this rendition segment", "streamer", did, "tfdt", tfdt) + log.Debug(ctx, "rendition playback: no source audio for this rendition segment", "tfdt", tfdt) continue } - name := names[id] + name := RenditionName(v.CodedWidth, v.CodedHeight) // The rendition's video plus the source's Opus tracks, in track-id // order, is a bare canonical segment, which Packetize presents and // selects from (the mix and the stems) like the source's own. @@ -161,9 +192,10 @@ func (mm *MediaManager) PublishRenditionsForPlayback(ctx context.Context, did st } packet, err := Packetize(ctx, mm.cli, &bus.Seg{Muxl: concatTracksByID(m4s)}) if err != nil { - log.Warn(ctx, "rendition playback: could not packetize rendition", "streamer", did, "rendition", name, "error", err) + log.Warn(ctx, "rendition playback: could not packetize rendition", "rendition", name, "error", err) continue } - mm.bus.PublishSegment(ctx, did, name, &bus.Seg{PacketizedData: packet, Published: published}) + packets[name] = packet } + return packets } -- 2.51.2