fix(hls): snapshot state in Store and publish fully initialized runs

Address the completion review findings on top of the upstream review
fixes:

- Store runs on the engine's checkpoint goroutine without the task
  lock and serializes the returned value afterwards: take the fetcher
  mutex and return a deep copy instead of the live state pointer,
  matching the HTTP protocol's Store pattern
- Load the journal and per-run base counters before publishing the run
  as active, so Stats/Progress never observe half-initialized fields
- Bind the journal fingerprint to the key IV as well: a rotated IV on
  an otherwise identical plan would silently decrypt to garbage
- Keep every EXT-X-MEDIA member per group instead of only the first
  one, so a group with a URI on a later member is still detected
- After a merge that raced with a pause, keep the staged segments and
  exit silently instead of cleaning them up; drop a dead context
  check and an unused helper, and make the MAP identity and merge
  part-file names unambiguous

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

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) {

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[:])
-}

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|")

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")