diff --git a/internal/protocol/hls/fetcher.go b/internal/protocol/hls/fetcher.go index 85e7ff40..7259a9b0 100644 --- a/internal/protocol/hls/fetcher.go +++ b/internal/protocol/hls/fetcher.go @@ -103,8 +103,9 @@ type hlsRun struct { downloaded atomic.Int64 doneCount atomic.Int64 failCount atomic.Int64 - baseBytes int64 // stored bytes restored from the journal at startup - baseDone int // completed segments restored from the journal at startup + baseBytes int64 // stored bytes restored from the journal at startup + baseDone int // completed segments restored from the journal at startup + pending []int64 // segment indexes to download, set by prepareRun } func (r *hlsRun) statsSnapshot() Stats { @@ -431,6 +432,12 @@ func (f *Fetcher) Start() error { if f.client == nil { f.client = f.buildClient() } + if f.meta.Res == nil { + // A restored plan without a resource cannot determine its output + // path; the engine re-resolves when Res is nil. + f.mu.Unlock() + return errors.New("hls fetcher is not resolved") + } ctx, cancel := context.WithCancel(context.Background()) run := &hlsRun{ @@ -447,17 +454,19 @@ func (f *Fetcher) Start() error { keyCache: make(map[string][]byte), journal: make(map[int64]int64), } + // Prepare staging and the completed-segment counters BEFORE publishing + // the run: Stats/Progress may observe it immediately afterwards and must + // never read half-initialized plain fields. + if err := f.prepareRun(run); err != nil { + f.mu.Unlock() + run.cancel() + close(run.done) + return err + } f.activeRun = run f.mu.Unlock() - if err := f.startRun(run); err != nil { - // No supervisor was started for this run, so close its done channel - // here to release anyone waiting on it. - run.cancel() - close(run.done) - f.clearActiveRun(run) - return err - } + f.launchWorkers(run) go f.supervise(run) return nil } @@ -563,12 +572,14 @@ func (f *Fetcher) supervise(run *hlsRun) { } else { err = f.merge(run) if err == nil { + if run.ctx.Err() != nil { + // Paused while the merge was in flight: keep the staged + // segments and exit silently; the next run re-merges. + f.clearActiveRun(run) + return + } os.RemoveAll(run.tempDir) } - if errors.Is(err, context.Canceled) { - f.clearActiveRun(run) - return - } } f.clearActiveRun(run) select { @@ -577,9 +588,11 @@ func (f *Fetcher) supervise(run *hlsRun) { } } -// startRun prepares the staging folder and launches the segment download -// workers. It returns immediately when every segment is already complete. -func (f *Fetcher) startRun(run *hlsRun) error { +// prepareRun prepares the staging folder and loads the completed-segment +// journal into run.baseBytes/baseDone/pending. It performs no goroutine +// launches and no context-bound work, so it runs safely under f.mu before +// the run is published. +func (f *Fetcher) prepareRun(run *hlsRun) error { if err := os.MkdirAll(run.tempDir, 0777); err != nil { return err } @@ -587,8 +600,6 @@ func (f *Fetcher) startRun(run *hlsRun) error { if err := loadJournal(run); err != nil { return err } - - var pending []int64 for i := range run.segs { if size, ok := run.journal[int64(i)]; ok { if info, err := os.Stat(segmentPath(run.tempDir, int64(i))); err == nil && info.Size() == size { @@ -597,19 +608,25 @@ func (f *Fetcher) startRun(run *hlsRun) error { continue } } - pending = append(pending, int64(i)) + run.pending = append(run.pending, int64(i)) } run.downloaded.Store(run.baseBytes) - if len(pending) == 0 { - return nil - } + return nil +} +// launchWorkers starts the queue producer and the segment download workers +// for a published run. Every goroutine registers itself in run.wg so Pause +// waits for all of them. +func (f *Fetcher) launchWorkers(run *hlsRun) { + if len(run.pending) == 0 { + return + } queue := make(chan int64) run.wg.Add(1) go func() { defer run.wg.Done() defer close(queue) - for _, idx := range pending { + for _, idx := range run.pending { select { case <-run.ctx.Done(): return @@ -618,7 +635,7 @@ func (f *Fetcher) startRun(run *hlsRun) error { } }() - workers := min(f.config.SegmentConnections, len(pending)) + workers := min(f.config.SegmentConnections, len(run.pending)) for i := 0; i < workers; i++ { run.wg.Add(1) go func() { @@ -641,7 +658,6 @@ func (f *Fetcher) startRun(run *hlsRun) error { } }() } - return nil } func (f *Fetcher) downloadSegment(run *hlsRun, idx int64) error { @@ -791,7 +807,7 @@ func (f *Fetcher) merge(run *hlsRun) error { if err := os.MkdirAll(filepath.Dir(target), 0777); err != nil { return err } - partPath := target + "." + tempDirPrefix + run.stagingID + ".part" + partPath := target + tempDirPrefix + run.stagingID + ".part" out, err := os.Create(partPath) if err != nil { return err diff --git a/internal/protocol/hls/fetcher_manager.go b/internal/protocol/hls/fetcher_manager.go index 5de89225..63941ec2 100644 --- a/internal/protocol/hls/fetcher_manager.go +++ b/internal/protocol/hls/fetcher_manager.go @@ -59,12 +59,26 @@ func (fm *FetcherManager) DefaultConfig() any { func (fm *FetcherManager) Store(f fetcher.Fetcher) (data any, err error) { _f := f.(*Fetcher) + _f.mu.Lock() + defer _f.mu.Unlock() if _f.state == nil { return nil, nil } - // The segment plan is immutable after resolve; the completed-segment - // journal lives on disk next to the segments. - return _f.state, nil + // The engine's periodic checkpoint calls Store without holding the task + // lock and serializes the returned value afterwards, while Resolve/Patch/ + // Start mutate the live state under the fetcher mutex: return a snapshot + // copy taken under that mutex, mirroring the HTTP protocol's Store. + return cloneState(_f.state), nil +} + +func cloneState(state *fetcherState) *fetcherState { + cloned := *state + cloned.Segments = make([]*Segment, len(state.Segments)) + for i, seg := range state.Segments { + s := *seg + cloned.Segments[i] = &s + } + return &cloned } func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) { diff --git a/internal/protocol/hls/helper.go b/internal/protocol/hls/helper.go index 7dd7c386..ba4a4f32 100644 --- a/internal/protocol/hls/helper.go +++ b/internal/protocol/hls/helper.go @@ -2,9 +2,7 @@ package hls import ( "context" - "crypto/sha1" "crypto/tls" - "encoding/hex" "fmt" "io" "net" @@ -100,9 +98,3 @@ func (r *stallReader) Read(p []byte) (int, error) { func (r *stallReader) Close() error { return r.rc.Close() } - -// sha1Hex returns the hex SHA-1 digest of s. -func sha1Hex(s string) string { - sum := sha1.Sum([]byte(s)) - return hex.EncodeToString(sum[:]) -} diff --git a/internal/protocol/hls/journal.go b/internal/protocol/hls/journal.go index 3de175a0..9d06a80b 100644 --- a/internal/protocol/hls/journal.go +++ b/internal/protocol/hls/journal.go @@ -47,7 +47,9 @@ func planFingerprint(state *fetcherState) string { fmt.Fprintf(h, "%d@%d|", seg.Byterange.Offset, seg.Byterange.Length) } if seg.Key != nil { - fmt.Fprintf(h, "key=%s|", seg.Key.URI) + // The IV is part of the plan: same URI with a rotated IV would + // silently decrypt to garbage if the journal were reused. + fmt.Fprintf(h, "key=%s iv=%x|", seg.Key.URI, seg.Key.IV) } if seg.IsInit { io.WriteString(h, "init|") diff --git a/internal/protocol/hls/playlist.go b/internal/protocol/hls/playlist.go index c62ab1e1..6dd1eeb2 100644 --- a/internal/protocol/hls/playlist.go +++ b/internal/protocol/hls/playlist.go @@ -32,7 +32,7 @@ type MediaRendition struct { // MasterPlaylist is a parsed master playlist with its rendition groups. type MasterPlaylist struct { Variants []*Variant - Rendition map[string]*MediaRendition // group id -> representative entry + Rendition map[string][]*MediaRendition // group id -> all EXT-X-MEDIA members } // Key is the decryption key applied to the segments that follow its tag, until @@ -111,7 +111,7 @@ func ParseMasterPlaylist(content string, baseURL *url.URL) (*MasterPlaylist, err if err := validateM3U8Header(content); err != nil { return nil, err } - master := &MasterPlaylist{Rendition: make(map[string]*MediaRendition)} + master := &MasterPlaylist{Rendition: make(map[string][]*MediaRendition)} var pending *Variant for _, line := range strings.Split(content, "\n") { line = strings.TrimSpace(line) @@ -142,11 +142,9 @@ func ParseMasterPlaylist(content string, baseURL *url.URL) (*MasterPlaylist, err rendition.URI = resolveURL(baseURL, uri) } if rendition.GroupID != "" { - // Keep the first entry per group: a group is unsupported as a - // whole when any of its members carries its own URI. - if _, ok := master.Rendition[rendition.GroupID]; !ok { - master.Rendition[rendition.GroupID] = rendition - } + // Keep every member: a group is unsupported as a whole when + // ANY of its members carries its own URI. + master.Rendition[rendition.GroupID] = append(master.Rendition[rendition.GroupID], rendition) } continue } @@ -172,13 +170,15 @@ func checkAudioRendition(master *MasterPlaylist, best *Variant) error { if best == nil || best.AudioGroup == "" { return nil } - rendition, ok := master.Rendition[best.AudioGroup] - if !ok || rendition == nil || rendition.Type != "AUDIO" || rendition.URI == "" { - return nil + for _, rendition := range master.Rendition[best.AudioGroup] { + if rendition == nil || rendition.Type != "AUDIO" || rendition.URI == "" { + continue + } + return fmt.Errorf( + "selected variant uses a separate audio rendition (group %q), which is not supported: the merged output would have no audio", + best.AudioGroup) } - return fmt.Errorf( - "selected variant uses a separate audio rendition (group %q), which is not supported: the merged output would have no audio", - best.AudioGroup) + return nil } // PickBestVariant selects the highest-bandwidth variant. @@ -256,8 +256,9 @@ func ParseMedia(content string, baseURL *url.URL) (*Media, error) { resolved := resolveURL(baseURL, uriAttr) // The init section identity covers every attribute, not just the // URI: a repeated tag with a different BYTERANGE addresses - // different bytes and changes the init section. - identity := resolved + "|" + attrs["BYTERANGE"] + // different bytes and changes the init section. NUL keeps the + // two parts unambiguously separated. + identity := resolved + "\x00" + attrs["BYTERANGE"] if mapIdentity != "" { if mapIdentity != identity { return nil, errors.New("changing init sections are not supported")