diff --git a/pkg/config/config.go b/pkg/config/config.go index fb7dcc3cb..5ab881ee0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -146,7 +146,6 @@ type CLI struct { WebsocketURL string BehindHTTPSProxy bool SegmentDebugDir string - LiveHLSDir string AdminDIDs []string Syndicate []string PlayerTelemetry bool @@ -753,12 +752,6 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { Destination: &cli.SegmentDebugDir, Sources: urfavecli.EnvVars("SP_SEGMENT_DEBUG_DIR"), }, - &urfavecli.StringFlag{ - Name: "live-hls-dir", - Usage: "if set, write a live MUXL fMP4 + HLS playlists per stream under this directory (opt-in; experimental)", - Destination: &cli.LiveHLSDir, - Sources: urfavecli.EnvVars("SP_LIVE_HLS_DIR"), - }, &urfavecli.StringFlag{ Name: "tickets", Usage: `tickets to join the swarm with (default: "")`, diff --git a/pkg/livehls/dir.go b/pkg/livehls/dir.go deleted file mode 100644 index b0ca469c7..000000000 --- a/pkg/livehls/dir.go +++ /dev/null @@ -1,35 +0,0 @@ -package livehls - -import ( - "os" - "path/filepath" -) - -// WriteHLSDir writes the current live HLS artifacts to dir: a master.m3u8, a -// per-track media playlist (.m3u8), and the per-track init segment -// (init-.mp4, the EXT-X-MAP target). The EXT-X-BYTERANGE entries -// point at blobName — the growing fMP4 the caller is writing alongside (e.g. -// "live.fmp4"). Call after each appended segment to refresh the live -// playlists; it's cheap (a few small files). -// -// The layout is self-contained and relative, so serving `dir` over plain HTTP -// (or syncing it to a bucket) is enough for an HLS player to follow the live -// stream. -func (w *Writer) WriteHLSDir(dir, blobName string) error { - if err := os.MkdirAll(dir, 0o755); err != nil { - return err - } - for _, tid := range w.TrackIDs() { - if init := w.InitSegment(tid); init != nil { - if err := os.WriteFile(filepath.Join(dir, "init-"+tid+".mp4"), init, 0o644); err != nil { - return err - } - } - pl := w.MediaPlaylist(tid, "init-"+tid+".mp4", blobName) - if err := os.WriteFile(filepath.Join(dir, tid+".m3u8"), []byte(pl), 0o644); err != nil { - return err - } - } - master := w.MasterPlaylist(func(tid string) string { return tid + ".m3u8" }) - return os.WriteFile(filepath.Join(dir, "master.m3u8"), []byte(master), 0o644) -} diff --git a/pkg/livehls/dir_test.go b/pkg/livehls/dir_test.go deleted file mode 100644 index 9d32a53b5..000000000 --- a/pkg/livehls/dir_test.go +++ /dev/null @@ -1,44 +0,0 @@ -package livehls - -import ( - "bytes" - "os" - "path/filepath" - "strings" - "testing" -) - -func TestWriteHLSDir(t *testing.T) { - var buf bytes.Buffer - w := NewWriter(&buf) - _ = w.Observe(initEvent()) - _ = w.Observe(segEvent(bytes.Repeat([]byte{1}, 100), bytes.Repeat([]byte{2}, 40))) - - dir := t.TempDir() - if err := w.WriteHLSDir(dir, "live.fmp4"); err != nil { - t.Fatalf("WriteHLSDir: %v", err) - } - - for _, name := range []string{"master.m3u8", "1.m3u8", "2.m3u8", "init-1.mp4", "init-2.mp4"} { - if _, err := os.Stat(filepath.Join(dir, name)); err != nil { - t.Errorf("expected %s to be written: %v", name, err) - } - } - - master, _ := os.ReadFile(filepath.Join(dir, "master.m3u8")) - if !strings.Contains(string(master), "#EXT-X-STREAM-INF") || !strings.Contains(string(master), "1.m3u8") { - t.Errorf("master.m3u8 missing video variant:\n%s", master) - } - - media, _ := os.ReadFile(filepath.Join(dir, "1.m3u8")) - for _, want := range []string{`#EXT-X-MAP:URI="init-1.mp4"`, "#EXT-X-BYTERANGE:", "live.fmp4"} { - if !strings.Contains(string(media), want) { - t.Errorf("1.m3u8 missing %q:\n%s", want, media) - } - } - - // init-1.mp4 should hold the per-track init bytes we supplied on the init event. - if init1, _ := os.ReadFile(filepath.Join(dir, "init-1.mp4")); string(init1) != "VINIT" { - t.Errorf("init-1.mp4 = %q, want VINIT", init1) - } -} diff --git a/pkg/livehls/livehls.go b/pkg/livehls/livehls.go index b052c5586..06033d66c 100644 --- a/pkg/livehls/livehls.go +++ b/pkg/livehls/livehls.go @@ -1,31 +1,35 @@ -// Package livehls is the "muxl hls for livestreams" write side: it turns the -// muxl event stream (init + canonical segments) into a growing, appendable -// fMP4 plus a per-track byte-range index, and serves live HLS playlists. +// Package livehls is the in-memory "muxl HLS for livestreams" window: it turns +// the muxl event stream (init + canonical segments) into a sliding window of +// recent segments held in memory, plus the HLS playlists that address them. // -// The model follows MUXL's presentation layering: the canonical segments are -// appended verbatim (so any C2PA/S2PA signature over a segment is preserved), -// and the index records each segment's absolute byte range within the fMP4. -// Playback is plain HLS — EXT-X-MAP for the init plus EXT-X-BYTERANGE per -// segment — with an increasing EXT-X-MEDIA-SEQUENCE and no EXT-X-ENDLIST until -// [Writer.Finalize] (which turns the live playlist into a VOD playlist). +// Nothing is written to the filesystem. The canonical .m4s segments are the +// durable unit (archived elsewhere, by ValidateMP4); this window only keeps +// the most recent few for live playback and serves them straight from RAM: +// +// - per-track init segment (ftyp+moov) — the EXT-X-MAP target +// - a windowed ring of recent canonical segments, addressed by a monotonic +// media-sequence number // -// fMP4 is the appendable format (the flat MP4 used for finalized VOD cannot be -// written incrementally), so this is the format a live stream uses. The index -// JSON shape matches pkg/vod.Metafile so the same read side can serve it. +// The segment bytes are appended verbatim, so any C2PA/S2PA signature over a +// segment is preserved and a player fetches exactly what was signed. Playback +// is plain fMP4 HLS — EXT-X-MAP for the init plus one URI per segment — with +// an advancing EXT-X-MEDIA-SEQUENCE and no EXT-X-ENDLIST until +// [Writer.Finalize] (which turns the live playlist into a VOD playlist). // -// Typical use, driven off muxl.RunMuxlSegmenter / RunMuxlSignSegment events: +// Typical use, driven off the muxl event stream a stream's segments arrive on +// (locally signed or replicated from another node): // -// w := livehls.NewWriter(fmp4File, livehls.WithWindow(6)) -// for ev := range eventCh { -// if err := w.Observe(ev); err != nil { ... } -// publish(w.MediaPlaylist(trackID, initURL, blobURL)) +// w := livehls.NewWriter(livehls.WithWindow(6)) +// w.Observe(initEvent) +// for ev := range segmentEvents { +// w.Observe(ev) // } -// w.Finalize() +// // serve w.MasterPlaylist(...), w.MediaPlaylist(...), w.InitSegment(tid), +// // w.SegmentData(tid, seq) over HTTP. package livehls import ( "fmt" - "io" "math" "sort" "strconv" @@ -35,126 +39,99 @@ import ( "stream.place/streamplace/pkg/muxl" ) -// Segment is one canonical segment's byte range within the growing fMP4. -// JSON tags match pkg/vod.MetafileSegment. +// Segment is one canonical segment held in the live window: a track's +// verbatim [c2pa?][muxl][moof][mdat] bytes plus the timing the playlist needs. type Segment struct { - Offset int64 `json:"offset"` - Size int64 `json:"size"` - DurationTicks uint64 `json:"durationTicks"` - SampleCount uint32 `json:"sampleCount"` + // Seq is the monotonic media-sequence number, assigned in arrival order + // and never reused — the value a player addresses the segment by and the + // basis for EXT-X-MEDIA-SEQUENCE. + Seq uint64 + DurationTicks uint64 + SampleCount uint32 + data []byte } -// Track is the per-track config plus its (windowed) segment index. JSON tags -// match pkg/vod.MetafileTrack so the existing playback read side can consume -// the index unchanged. +// Size is the segment's byte length. +func (s Segment) Size() int { return len(s.data) } + +// Track is the per-track config plus its (windowed) in-memory segments. type Track struct { - Type string `json:"type"` - Codec string `json:"codec"` - Timescale uint32 `json:"timescale"` - Segments []Segment `json:"segments"` - Width uint32 `json:"width,omitempty"` - Height uint32 `json:"height,omitempty"` - Channels uint32 `json:"channels,omitempty"` - SampleRate uint32 `json:"sampleRate,omitempty"` - - // init is the per-track ftyp+moov (EXT-X-MAP target); not serialized. - init []byte - // mediaSeq is the number of segments evicted from the head of the - // playlist window — i.e. EXT-X-MEDIA-SEQUENCE. - mediaSeq uint64 -} + Type string + Codec string + Timescale uint32 + Width uint32 + Height uint32 + Channels uint32 + SampleRate uint32 -// InitSegment returns the per-track init (ftyp+moov) bytes for trackID, the -// EXT-X-MAP target, or nil if unknown. -func (w *Writer) InitSegment(trackID string) []byte { - w.mu.Lock() - defer w.mu.Unlock() - if t := w.tracks[trackID]; t != nil { - return t.init - } - return nil + Segments []Segment + + // init is the per-track ftyp+moov (EXT-X-MAP target); not part of the + // window. nextSeq is the next media-sequence number to hand out. + init []byte + nextSeq uint64 } -// Writer assembles a live MUXL fMP4 and its HLS index incrementally. It is -// safe for concurrent Observe and playlist reads. +// Writer assembles an in-memory live HLS window from the muxl event stream. +// Safe for concurrent Observe and read/serve. type Writer struct { mu sync.Mutex - out io.Writer - running int64 // bytes appended to out so far (= next segment offset) tracks map[string]*Track - order []string // track ids, first-seen order - window int // max segments per media playlist; 0 = keep all + order []string // track ids, sorted + window int // max segments retained per track; 0 = keep all finished bool } // Option configures a Writer. type Option func(*Writer) -// WithWindow keeps at most n segments per track in the live playlist (a -// sliding window), advancing EXT-X-MEDIA-SEQUENCE as older segments fall off. -// The underlying fMP4 bytes are not truncated; only the playlist window -// slides. n <= 0 keeps every segment (an "event" playlist). +// WithWindow keeps at most n segments per track in memory (a sliding window), +// advancing EXT-X-MEDIA-SEQUENCE as older segments are evicted. n <= 0 keeps +// every segment (an "event"/VOD-style window — unbounded memory, avoid for +// 24/7 streams). func WithWindow(n int) Option { return func(w *Writer) { w.window = n } } -// NewWriter returns a live HLS writer that appends the init and segments to -// out (a file, blob multipart writer, etc.). -func NewWriter(out io.Writer, opts ...Option) *Writer { - w := &Writer{out: out, tracks: map[string]*Track{}} +// NewWriter returns an empty in-memory live HLS window. +func NewWriter(opts ...Option) *Writer { + w := &Writer{tracks: map[string]*Track{}} for _, opt := range opts { opt(w) } return w } -// Observe appends one muxl event to the fMP4 and updates the index. Events -// must arrive in stream order (init first, then segments in emission order), -// since byte offsets are computed from the cumulative bytes written. +// Observe folds one muxl event into the window. An "init" event supplies the +// per-track init segments and catalog; "segment"/"signed-segment" events +// append each track's canonical-segment bytes (copied — the caller may reuse +// its buffer) and evict beyond the window. func (w *Writer) Observe(ev *muxl.MuxlEvent) error { w.mu.Lock() defer w.mu.Unlock() switch ev.Type { case "init": - // The combined ftyp+moov is written once at the head of the fMP4 so - // the file is a valid, self-contained MUXL fMP4; players use the - // per-track EXT-X-MAP init and byte-range into the segments, skipping - // this header. - if _, err := w.out.Write(ev.Data); err != nil { - return fmt.Errorf("livehls: write init: %w", err) - } - w.running += int64(len(ev.Data)) for tid, initBytes := range ev.TrackInits { - t := w.track(tid) - t.init = append([]byte(nil), initBytes...) + w.track(tid).init = append([]byte(nil), initBytes...) } if ev.Catalog != nil { w.applyCatalog(ev.Catalog) } case "segment", "signed-segment": - // Per-track chunks are concatenated in sorted track-id order, matching - // the byte layout muxl emits; track that order so offsets line up. A - // signed-segment chunk carries a leading c2pa-uuid box, which only - // changes its length — the offset math is unchanged. for _, tid := range sortedKeys(ev.Tracks) { - chunk := ev.Tracks[tid] - if _, err := w.out.Write(chunk); err != nil { - return fmt.Errorf("livehls: write segment (track %s): %w", tid, err) - } t := w.track(tid) t.Segments = append(t.Segments, Segment{ - Offset: w.running, - Size: int64(len(chunk)), + Seq: t.nextSeq, DurationTicks: ev.Durations[tid], SampleCount: ev.SampleCounts[tid], + data: append([]byte(nil), ev.Tracks[tid]...), }) - w.running += int64(len(chunk)) + t.nextSeq++ if w.window > 0 && len(t.Segments) > w.window { drop := len(t.Segments) - w.window t.Segments = append(t.Segments[:0:0], t.Segments[drop:]...) - t.mediaSeq += uint64(drop) } } @@ -172,14 +149,15 @@ func (w *Writer) Finalize() { w.finished = true } -// TrackIDs returns the known track ids in first-seen order. +// TrackIDs returns the known track ids in sorted order. func (w *Writer) TrackIDs() []string { w.mu.Lock() defer w.mu.Unlock() return append([]string(nil), w.order...) } -// Track returns a snapshot of the track's current index, or nil if unknown. +// Track returns a snapshot of the track's current windowed segments (segment +// byte payloads are shared, not copied), or nil if unknown. func (w *Writer) Track(trackID string) *Track { w.mu.Lock() defer w.mu.Unlock() @@ -193,11 +171,38 @@ func (w *Writer) Track(trackID string) *Track { return &cp } +// InitSegment returns the per-track init (ftyp+moov) bytes — the EXT-X-MAP +// target — or nil if unknown. +func (w *Writer) InitSegment(trackID string) []byte { + w.mu.Lock() + defer w.mu.Unlock() + if t := w.tracks[trackID]; t != nil { + return t.init + } + return nil +} + +// SegmentData returns the bytes of the segment with media-sequence seq for +// trackID, or nil if it's unknown or has already slid out of the window. +func (w *Writer) SegmentData(trackID string, seq uint64) []byte { + w.mu.Lock() + defer w.mu.Unlock() + t := w.tracks[trackID] + if t == nil { + return nil + } + for _, s := range t.Segments { + if s.Seq == seq { + return s.data + } + } + return nil +} + // MediaPlaylist renders the live HLS media playlist for trackID. initURL is -// the EXT-X-MAP target (the per-track init); blobURL is the URL of the growing -// fMP4 that the EXT-X-BYTERANGE entries address. Returns "" for an unknown -// track. -func (w *Writer) MediaPlaylist(trackID, initURL, blobURL string) string { +// the EXT-X-MAP target (the per-track init); segURI maps a segment's +// media-sequence number to its URI. Returns "" for an unknown track. +func (w *Writer) MediaPlaylist(trackID, initURL string, segURI func(seq uint64) string) string { w.mu.Lock() defer w.mu.Unlock() t := w.tracks[trackID] @@ -215,18 +220,22 @@ func (w *Writer) MediaPlaylist(trackID, initURL, blobURL string) string { if target < 1 { target = 1 } + mediaSeq := uint64(0) + if len(t.Segments) > 0 { + mediaSeq = t.Segments[0].Seq + } var b strings.Builder b.WriteString("#EXTM3U\n#EXT-X-VERSION:7\n") fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", target) - fmt.Fprintf(&b, "#EXT-X-MEDIA-SEQUENCE:%d\n", t.mediaSeq) + fmt.Fprintf(&b, "#EXT-X-MEDIA-SEQUENCE:%d\n", mediaSeq) b.WriteString("#EXT-X-INDEPENDENT-SEGMENTS\n") if w.finished { b.WriteString("#EXT-X-PLAYLIST-TYPE:VOD\n") } fmt.Fprintf(&b, "#EXT-X-MAP:URI=%q\n", initURL) for _, s := range t.Segments { - fmt.Fprintf(&b, "#EXTINF:%.6f,\n#EXT-X-BYTERANGE:%d@%d\n%s\n", s.seconds(t.Timescale), s.Size, s.Offset, blobURL) + fmt.Fprintf(&b, "#EXTINF:%.6f,\n%s\n", s.seconds(t.Timescale), segURI(s.Seq)) } if w.finished { b.WriteString("#EXT-X-ENDLIST\n") @@ -290,8 +299,7 @@ func (w *Writer) track(tid string) *Track { return t } -// applyCatalog fills per-track config from the muxl catalog, mirroring -// vod.metafileBuilder.Finalize. +// applyCatalog fills per-track config from the muxl catalog. func (w *Writer) applyCatalog(cat *muxl.MuxlCatalog) { if cat.Video != nil { for _, c := range cat.Video.Renditions { @@ -327,7 +335,7 @@ func (t *Track) peakBitrate() int { var bytes int64 var ticks uint64 for _, s := range t.Segments { - bytes += s.Size + bytes += int64(s.Size()) ticks += s.DurationTicks } if ticks == 0 || t.Timescale == 0 { diff --git a/pkg/livehls/livehls_test.go b/pkg/livehls/livehls_test.go index e81d6d587..4e6ae16fe 100644 --- a/pkg/livehls/livehls_test.go +++ b/pkg/livehls/livehls_test.go @@ -2,6 +2,7 @@ package livehls import ( "bytes" + "fmt" "strings" "testing" @@ -33,9 +34,10 @@ func segEvent(v, a []byte) *muxl.MuxlEvent { } } -func TestWriterAppendsVerbatimAndIndexes(t *testing.T) { - var buf bytes.Buffer - w := NewWriter(&buf) +func segURI(seq uint64) string { return fmt.Sprintf("seg%d.m4s", seq) } + +func TestWriterStoresSegmentsInMemory(t *testing.T) { + w := NewWriter() if err := w.Observe(initEvent()); err != nil { t.Fatal(err) } @@ -50,47 +52,50 @@ func TestWriterAppendsVerbatimAndIndexes(t *testing.T) { t.Fatal(err) } - // fMP4 = INIT + (v1,a1) + (v2,a2): per-segment tracks in sorted id order. - want := bytes.Join([][]byte{[]byte("INIT"), v1, a1, v2, a2}, nil) - if !bytes.Equal(buf.Bytes(), want) { - t.Fatalf("fmp4 bytes mismatch: got %d bytes, want %d", buf.Len(), len(want)) - } - - // Track 1 (video): seg0 @ len("INIT")=4, seg1 @ 4+100+40=144. + // Track 1 (video): two segments, media sequences 0 and 1. tr1 := w.Track("1") if tr1.Type != "video" || tr1.Width != 1280 || tr1.Height != 720 { t.Errorf("track1 config wrong: %+v", tr1) } - if len(tr1.Segments) != 2 || - tr1.Segments[0] != (Segment{Offset: 4, Size: 100, DurationTicks: 90000, SampleCount: 30}) || - tr1.Segments[1] != (Segment{Offset: 144, Size: 110, DurationTicks: 90000, SampleCount: 30}) { - t.Errorf("track1 segments wrong: %+v", tr1.Segments) + if len(tr1.Segments) != 2 { + t.Fatalf("track1 should have 2 segments, got %d", len(tr1.Segments)) + } + if tr1.Segments[0].Seq != 0 || tr1.Segments[0].Size() != 100 || + tr1.Segments[0].DurationTicks != 90000 || tr1.Segments[0].SampleCount != 30 { + t.Errorf("track1 seg0 wrong: %+v", tr1.Segments[0]) + } + if tr1.Segments[1].Seq != 1 || tr1.Segments[1].Size() != 110 { + t.Errorf("track1 seg1 wrong: %+v", tr1.Segments[1]) + } + // Segment bytes are served verbatim from memory, addressed by media seq. + if !bytes.Equal(w.SegmentData("1", 0), v1) || !bytes.Equal(w.SegmentData("1", 1), v2) { + t.Errorf("track1 SegmentData did not return the verbatim segment bytes") + } + if string(w.InitSegment("1")) != "VINIT" { + t.Errorf("track1 init = %q, want VINIT", w.InitSegment("1")) } - // Track 2 (audio): seg0 @ 4+100=104, seg1 @ 144+110=254. + // Track 2 (audio). tr2 := w.Track("2") if tr2.Type != "audio" || tr2.SampleRate != 48000 || tr2.Channels != 2 { t.Errorf("track2 config wrong: %+v", tr2) } - if tr2.Segments[0].Offset != 104 || tr2.Segments[0].Size != 40 || - tr2.Segments[1].Offset != 254 || tr2.Segments[1].Size != 42 { - t.Errorf("track2 segments wrong: %+v", tr2.Segments) + if !bytes.Equal(w.SegmentData("2", 0), a1) || !bytes.Equal(w.SegmentData("2", 1), a2) { + t.Errorf("track2 SegmentData did not return the verbatim segment bytes") } } func TestMediaPlaylistLiveThenFinalize(t *testing.T) { - var buf bytes.Buffer - w := NewWriter(&buf) + w := NewWriter() _ = w.Observe(initEvent()) _ = w.Observe(segEvent(bytes.Repeat([]byte{1}, 100), bytes.Repeat([]byte{2}, 40))) - pl := w.MediaPlaylist("1", "init1.mp4", "blob.fmp4") + pl := w.MediaPlaylist("1", "init1.mp4", segURI) for _, want := range []string{ `#EXT-X-MAP:URI="init1.mp4"`, "#EXT-X-MEDIA-SEQUENCE:0", "#EXTINF:1.000000,", - "#EXT-X-BYTERANGE:100@4", - "blob.fmp4", + "seg0.m4s", } { if !strings.Contains(pl, want) { t.Errorf("live playlist missing %q:\n%s", want, pl) @@ -101,36 +106,40 @@ func TestMediaPlaylistLiveThenFinalize(t *testing.T) { } w.Finalize() - pl = w.MediaPlaylist("1", "init1.mp4", "blob.fmp4") + pl = w.MediaPlaylist("1", "init1.mp4", segURI) if !strings.Contains(pl, "#EXT-X-ENDLIST") || !strings.Contains(pl, "#EXT-X-PLAYLIST-TYPE:VOD") { t.Errorf("finalized playlist must be a VOD playlist with ENDLIST:\n%s", pl) } } func TestSlidingWindowEvictsAndAdvancesMediaSequence(t *testing.T) { - var buf bytes.Buffer - w := NewWriter(&buf, WithWindow(2)) + w := NewWriter(WithWindow(2)) _ = w.Observe(initEvent()) for i := 0; i < 4; i++ { _ = w.Observe(segEvent([]byte{byte(i)}, []byte{byte(i)})) } tr := w.Track("1") if len(tr.Segments) != 2 { - t.Fatalf("window=2 should cap the playlist at 2 segments, got %d", len(tr.Segments)) + t.Fatalf("window=2 should cap the window at 2 segments, got %d", len(tr.Segments)) } - // 4 emitted, 2 retained → 2 evicted → media sequence 2. - if !strings.Contains(w.MediaPlaylist("1", "i", "b"), "#EXT-X-MEDIA-SEQUENCE:2") { + // 4 emitted, 2 retained → media sequences 2 and 3 remain. + if tr.Segments[0].Seq != 2 || tr.Segments[1].Seq != 3 { + t.Errorf("expected retained seqs [2,3], got [%d,%d]", tr.Segments[0].Seq, tr.Segments[1].Seq) + } + if !strings.Contains(w.MediaPlaylist("1", "i", segURI), "#EXT-X-MEDIA-SEQUENCE:2") { t.Errorf("expected EXT-X-MEDIA-SEQUENCE:2 after evicting 2 segments") } - // Bytes are never truncated: all 4 segments (both tracks) plus init remain. - if buf.Len() != len("INIT")+4*2 { - t.Errorf("underlying fmp4 should retain all bytes, got %d", buf.Len()) + // Evicted segments are gone from memory; retained ones are served. + if w.SegmentData("1", 0) != nil { + t.Errorf("evicted segment 0 should no longer be retained") + } + if w.SegmentData("1", 3) == nil { + t.Errorf("segment 3 should still be retained") } } func TestMasterPlaylist(t *testing.T) { - var buf bytes.Buffer - w := NewWriter(&buf) + w := NewWriter() _ = w.Observe(initEvent()) _ = w.Observe(segEvent(bytes.Repeat([]byte{1}, 100), bytes.Repeat([]byte{2}, 40))) diff --git a/pkg/livehls/run.go b/pkg/livehls/run.go index b4576460f..c4c3c1ba6 100644 --- a/pkg/livehls/run.go +++ b/pkg/livehls/run.go @@ -7,27 +7,24 @@ import ( "stream.place/streamplace/pkg/muxl" ) -// Run drives the muxl segmenter over an fMP4 input and assembles an unsigned -// live HLS presentation: every muxl event is fed into a Writer, which appends -// the init and each canonical segment to out (a growing fMP4) and maintains -// the per-track byte-range index. It returns the populated Writer once the -// input is fully consumed — call its MediaPlaylist/MasterPlaylist/Track -// methods to serve playback. -func Run(ctx context.Context, input io.Reader, out io.Writer, opts ...Option) (*Writer, error) { - return drive(NewWriter(out, opts...), func(eventCh chan *muxl.MuxlEvent) error { +// Run drives the muxl segmenter over an fMP4 input and folds every event into +// an in-memory live HLS window, returning the populated Writer once the input +// is fully consumed — serve its MediaPlaylist/MasterPlaylist/InitSegment/ +// SegmentData for playback. +func Run(ctx context.Context, input io.Reader, opts ...Option) (*Writer, error) { + return drive(NewWriter(opts...), func(eventCh chan *muxl.MuxlEvent) error { return muxl.RunMuxlSegmenterEvents(ctx, input, eventCh) }) } // RunSigned is the signed counterpart of Run: it drives muxl-sign's -// sign-segment over the fMP4 input, so each canonical segment appended to out -// is C2PA/S2PA-signed ([c2pa-uuid][muxl-uuid][moof][mdat]). The byte-range -// index and playlists are identical in shape to Run — the only difference is -// the segment bytes carry the signing prefix, which is included in the range. -// This is the live-HLS shape for a signed broadcast. in.KeyPEM (+ CertPEM and -// manifests) must be set, matching muxl.RunMuxlSignSegment. -func RunSigned(ctx context.Context, input io.Reader, in muxl.SignerInput, out io.Writer, opts ...Option) (*Writer, error) { - return drive(NewWriter(out, opts...), func(eventCh chan *muxl.MuxlEvent) error { +// sign-segment over the fMP4 input, so each canonical segment folded into the +// window is C2PA/S2PA-signed ([c2pa-uuid][muxl-uuid][moof][mdat]). The +// playlists are identical in shape to Run; the only difference is the segment +// bytes carry the signing prefix. in.KeyPEM (or in.Sign) plus CertPEM and +// manifests must be set, matching muxl.RunMuxlSignSegment. +func RunSigned(ctx context.Context, input io.Reader, in muxl.SignerInput, opts ...Option) (*Writer, error) { + return drive(NewWriter(opts...), func(eventCh chan *muxl.MuxlEvent) error { return muxl.RunMuxlSignSegment(ctx, input, in, nil, nil, eventCh) }) } diff --git a/pkg/livehls/run_test.go b/pkg/livehls/run_test.go index 1fd39f1f7..51bfb943c 100644 --- a/pkg/livehls/run_test.go +++ b/pkg/livehls/run_test.go @@ -13,8 +13,8 @@ import ( ) // TestRunProducesLiveHLSFromFmp4 drives a real fMP4 through the muxl wasm -// segmenter and the live HLS writer, then checks that the growing fMP4 and -// the per-track byte-range index/playlists are internally consistent. +// segmenter into the in-memory live HLS window, then checks the per-track +// segments and playlists are internally consistent. func TestRunProducesLiveHLSFromFmp4(t *testing.T) { _, thisFile, _, _ := runtime.Caller(0) repoRoot := filepath.Join(filepath.Dir(thisFile), "..", "..") @@ -23,20 +23,11 @@ func TestRunProducesLiveHLSFromFmp4(t *testing.T) { t.Fatalf("read fixture: %v", err) } - var out bytes.Buffer - w, err := Run(context.Background(), bytes.NewReader(fmp4), &out) + w, err := Run(context.Background(), bytes.NewReader(fmp4)) if err != nil { t.Fatalf("Run: %v", err) } - if out.Len() == 0 { - t.Fatal("no fMP4 bytes written") - } - // The growing fMP4 must begin with the init segment (ftyp box). - if out.Len() < 8 || string(out.Bytes()[4:8]) != "ftyp" { - t.Fatalf("output fMP4 should start with an ftyp box, got %q", firstFourCC(out.Bytes())) - } - tids := w.TrackIDs() if len(tids) == 0 { t.Fatal("no tracks produced") @@ -46,25 +37,26 @@ func TestRunProducesLiveHLSFromFmp4(t *testing.T) { for _, tid := range tids { tr := w.Track(tid) for _, s := range tr.Segments { - // Every indexed byte range must lie within the bytes we appended. - if s.Size <= 0 || s.Offset < int64(len("ftyp")) || s.Offset+s.Size > int64(out.Len()) { - t.Errorf("track %s segment byte range out of bounds: %+v (fmp4 %d bytes)", tid, s, out.Len()) + if s.Size() <= 0 { + t.Errorf("track %s segment %d is empty", tid, s.Seq) + } + if w.SegmentData(tid, s.Seq) == nil { + t.Errorf("track %s segment %d not retrievable by media sequence", tid, s.Seq) } totalSegs++ } - pl := w.MediaPlaylist(tid, "init.mp4", "blob.fmp4") - for _, want := range []string{"#EXTM3U", "#EXT-X-MAP:", "#EXT-X-BYTERANGE:"} { + pl := w.MediaPlaylist(tid, "init.mp4", segURI) + for _, want := range []string{"#EXTM3U", "#EXT-X-MAP:", "#EXTINF:"} { if !strings.Contains(pl, want) { t.Errorf("track %s media playlist missing %q:\n%s", tid, want, pl) } } - // Live (not finalized) → no ENDLIST yet. if strings.Contains(pl, "#EXT-X-ENDLIST") { t.Errorf("live playlist should not have ENDLIST before Finalize") } } if totalSegs == 0 { - t.Fatal("no segments indexed") + t.Fatal("no segments produced") } m := w.MasterPlaylist(func(tid string) string { return "t" + tid + ".m3u8" }) @@ -72,20 +64,13 @@ func TestRunProducesLiveHLSFromFmp4(t *testing.T) { t.Errorf("master playlist malformed:\n%s", m) } - t.Logf("live HLS: %d tracks, %d segments, %d fMP4 bytes", len(tids), totalSegs, out.Len()) -} - -func firstFourCC(b []byte) string { - if len(b) >= 8 { - return string(b[4:8]) - } - return "" + t.Logf("live HLS: %d tracks, %d segments", len(tids), totalSegs) } // TestRunSignedProducesSignedLiveHLS drives the same fixture through the -// signing segmenter and checks that each appended segment is a signed m4s — -// i.e. its byte range begins with a uuid box (the c2pa/S2PA prefix). Skips if -// the sibling muxl repo's test keys aren't present. +// signing segmenter and checks each windowed segment is a signed m4s — i.e. +// its bytes begin with a uuid box (the c2pa/S2PA prefix). Skips if the sibling +// muxl repo's test keys aren't present. func TestRunSignedProducesSignedLiveHLS(t *testing.T) { _, thisFile, _, _ := runtime.Caller(0) repoRoot := filepath.Join(filepath.Dir(thisFile), "..", "..") @@ -106,27 +91,26 @@ func TestRunSignedProducesSignedLiveHLS(t *testing.T) { manifest := []byte(`{"title":"livehls signed test","assertions":[{"label":"c2pa.actions","data":{"actions":[{"action":"c2pa.created"}]}}]}`) - var out bytes.Buffer w, err := RunSigned(context.Background(), bytes.NewReader(fmp4), muxl.SignerInput{ CertPEM: certPEM, KeyPEM: keyPEM, TrackManifest: manifest, WrapperManifest: manifest, - }, &out) + }) if err != nil { t.Fatalf("RunSigned: %v", err) } - body := out.Bytes() signedSegs := 0 for _, tid := range w.TrackIDs() { for _, s := range w.Track(tid).Segments { - if s.Offset+s.Size > int64(len(body)) { - t.Fatalf("segment out of range: %+v (%d bytes)", s, len(body)) + data := w.SegmentData(tid, s.Seq) + if len(data) < 8 { + t.Fatalf("track %s segment %d too short: %d bytes", tid, s.Seq, len(data)) } // A signed canonical segment leads with the c2pa uuid box. - if got := string(body[s.Offset+4 : s.Offset+8]); got != "uuid" { - t.Errorf("track %s segment should start with a uuid box, got %q", tid, got) + if got := string(data[4:8]); got != "uuid" { + t.Errorf("track %s segment %d should start with a uuid box, got %q", tid, s.Seq, got) } signedSegs++ } @@ -134,5 +118,5 @@ func TestRunSignedProducesSignedLiveHLS(t *testing.T) { if signedSegs == 0 { t.Fatal("no signed segments produced") } - t.Logf("signed live HLS: %d segments, %d fMP4 bytes", signedSegs, len(body)) + t.Logf("signed live HLS: %d segments", signedSegs) } diff --git a/pkg/media/muxl_segment.go b/pkg/media/muxl_segment.go index f9eb3d8f4..342a719a1 100644 --- a/pkg/media/muxl_segment.go +++ b/pkg/media/muxl_segment.go @@ -3,19 +3,13 @@ package media import ( "bytes" "context" - _ "embed" "fmt" "io" - "os" - "path/filepath" - "sort" - "strings" "time" "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "stream.place/streamplace/pkg/config" - "stream.place/streamplace/pkg/livehls" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/muxl" ) @@ -86,48 +80,40 @@ func MuxlSegmentElem(ctx context.Context, cli *config.CLI, streamer string, doH2 r.Close() }() - if cli.LiveHLSDir != "" { - // Opt-in (experimental): drive the new rich-event segmenter and feed - // a live HLS writer. The per-GoP fMP4 delivered to cb is reconstructed - // to be byte-identical to the default path (init + sorted-track-concat), - // so signing/validation downstream is unaffected. - go runMuxlLiveHLS(ctx, cli, streamer, r, cb) - } else { - initCh := make(chan []byte) - segCh := make(chan []byte) - go func() { - err := muxl.RunMuxlSegmenter(ctx, r, initCh, segCh) - if err != nil { - log.Error(ctx, "error running muxl segmenter", "error", err) - } - }() + initCh := make(chan []byte) + segCh := make(chan []byte) + go func() { + err := muxl.RunMuxlSegmenter(ctx, r, initCh, segCh) + if err != nil { + log.Error(ctx, "error running muxl segmenter", "error", err) + } + }() - go func() { - var initSeg []byte + go func() { + var initSeg []byte + select { + case <-ctx.Done(): + return + case initSeg = <-initCh: + log.Debug(ctx, "got init segment", "size", len(initSeg)) + } + for { select { case <-ctx.Done(): return - case initSeg = <-initCh: - log.Debug(ctx, "got init segment", "size", len(initSeg)) - } - for { - select { - case <-ctx.Done(): - return - case seg := <-segCh: - log.Debug(ctx, "got segment", "size", len(seg)) - fullSeg := []byte{} - fullSeg = append(fullSeg, initSeg...) - fullSeg = append(fullSeg, seg...) - cli.DumpDebugSegment(ctx, "muxl_segment_input.fmp4", bytes.NewReader(fullSeg)) - err := cb(ctx, fullSeg, time.Now().UnixMilli()) - if err != nil { - log.Error(ctx, "error calling callback", "error", err) - } + case seg := <-segCh: + log.Debug(ctx, "got segment", "size", len(seg)) + fullSeg := []byte{} + fullSeg = append(fullSeg, initSeg...) + fullSeg = append(fullSeg, seg...) + cli.DumpDebugSegment(ctx, "muxl_segment_input.fmp4", bytes.NewReader(fullSeg)) + err := cb(ctx, fullSeg, time.Now().UnixMilli()) + if err != nil { + log.Error(ctx, "error calling callback", "error", err) } } - }() - } + } + }() sink := app.SinkFromElement(appsink) sink.SetCallbacks(&app.SinkCallbacks{ @@ -136,81 +122,3 @@ func MuxlSegmentElem(ctx context.Context, cli *config.CLI, streamer string, doH2 return bin.Element, nil } - -// runMuxlLiveHLS consumes the muxl segmenter's rich event stream from r and -// (a) reconstructs the per-GoP fMP4 the default path delivers to cb — init + -// the per-track segment bytes concatenated in sorted track-id order, exactly -// as ParseMuxlEvents/segCh does — and (b) feeds every event into a live HLS -// writer, appending to a growing fMP4 and refreshing the playlists under -// cli.LiveHLSDir//. Best-effort: live HLS errors are logged, never -// fatal to the (signing) callback path. -func runMuxlLiveHLS(ctx context.Context, cli *config.CLI, streamer string, r io.Reader, cb func(context.Context, []byte, int64) error) { - streamDir := filepath.Join(cli.LiveHLSDir, fileSafeName(streamer)) - if err := os.MkdirAll(streamDir, 0o755); err != nil { - log.Error(ctx, "live-hls: mkdir failed", "dir", streamDir, "error", err) - return - } - f, err := os.Create(filepath.Join(streamDir, "live.fmp4")) - if err != nil { - log.Error(ctx, "live-hls: create fmp4 failed", "error", err) - return - } - defer f.Close() - lw := livehls.NewWriter(f) - - eventCh := make(chan *muxl.MuxlEvent, 16) - errCh := make(chan error, 1) - go func() { - err := muxl.RunMuxlSegmenterEvents(ctx, r, eventCh) - close(eventCh) - errCh <- err - }() - - var initSeg []byte - for ev := range eventCh { - if err := lw.Observe(ev); err != nil { - log.Error(ctx, "live-hls: observe failed", "error", err) - } - switch ev.Type { - case "init": - initSeg = ev.Data - log.Debug(ctx, "got init segment", "size", len(initSeg)) - case "segment", "signed-segment": - seg := concatTracksSorted(ev.Tracks) - fullSeg := make([]byte, 0, len(initSeg)+len(seg)) - fullSeg = append(fullSeg, initSeg...) - fullSeg = append(fullSeg, seg...) - cli.DumpDebugSegment(ctx, "muxl_segment_input.fmp4", bytes.NewReader(fullSeg)) - if err := cb(ctx, fullSeg, time.Now().UnixMilli()); err != nil { - log.Error(ctx, "error calling callback", "error", err) - } - if err := lw.WriteHLSDir(streamDir, "live.fmp4"); err != nil { - log.Error(ctx, "live-hls: write playlists failed", "error", err) - } - } - } - if err := <-errCh; err != nil && ctx.Err() == nil { - log.Error(ctx, "error running muxl segmenter (live-hls)", "error", err) - } -} - -// concatTracksSorted joins per-track segment bytes in sorted track-id order, -// matching muxl.ParseMuxlEvents' segCh byte layout so the reconstructed fMP4 -// is identical to the default path's. -func concatTracksSorted(tracks map[string][]byte) []byte { - keys := make([]string, 0, len(tracks)) - for k := range tracks { - keys = append(keys, k) - } - sort.Strings(keys) - var out []byte - for _, k := range keys { - out = append(out, tracks[k]...) - } - return out -} - -// fileSafeName makes a streamer identifier safe to use as a directory name. -func fileSafeName(s string) string { - return strings.NewReplacer("/", "-", ":", "-", "\\", "-", " ", "_").Replace(s) -}