diff --git a/cmd/magnetowid/main.go b/cmd/magnetowid/main.go index ce8c260..4683b2a 100644 --- a/cmd/magnetowid/main.go +++ b/cmd/magnetowid/main.go @@ -106,7 +106,7 @@ func run(log *slog.Logger, level *slog.LevelVar) error { return err } - queue, err := downloader.New(dir, db, providers, &downloader.FFmpeg{Path: *ffmpeg}, log) + queue, err := downloader.New(dir, db, providers, &downloader.FFmpeg{Path: *ffmpeg, Log: log}, log) if err != nil { return err } diff --git a/docs/usage.md b/docs/usage.md index 98c4f71..783fc6b 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -159,8 +159,8 @@ curl 'http://localhost:8484/api?mode=queue&name=resume&value=JOB_ID&apikey=YOUR_ ``` `mode=queue&apikey=YOUR_API_KEY` lists job IDs (`nzo_id`). Use `name=pause` -and comma-separated IDs in `value` to pause jobs. Pauses survive restarts; -interrupted downloads restart from the beginning when resumed. +and comma-separated IDs in `value` to pause jobs. Pauses survive restarts, and +resumed downloads continue where they stopped. ## Web interface @@ -168,9 +168,9 @@ Open `http://:8484/` in a browser and sign in with the API key. **Queue** shows the running download's progress and time left, queued jobs in run order, and paused or retrying jobs with their last error. It refreshes -every second. Pause or resume the queue or a job, or remove a job; removing the -running download deletes its partial file. Resumed downloads restart from the -beginning. +every second. Pause or resume the queue or a job, or remove a job; removing an +unfinished download deletes its partial files. Resumed downloads continue where +they stopped. **History** lists finished and failed downloads, newest first, and refreshes every five seconds. Downloads stay for 30 days unless Sonarr or Radarr removes @@ -301,9 +301,15 @@ Include the steps to reproduce and any relevant error message, with API keys rem known audio language. Probes run during search and are cached for a day. Downloads keep one audio track; subtitle conversion supports TTML only. - **Restarts:** `.magnetowid-jobs.db` stores the queue, history, and watch lists. - Completed history expires after 30 days. Interrupted downloads restart from - the beginning. The download filesystem must support file locks; only one - magnetowid instance can use it at a time. + Completed history expires after 30 days. The download filesystem must support + file locks; only one magnetowid instance can use it at a time. +- **Partial downloads:** magnetowid fetches HLS segments, four at a time, into + `.incomplete/` in the download folder, then remuxes them with ffmpeg. + Paused, interrupted and retried downloads continue from their last complete + segment, unless the site now serves a different stream. Other streams, such + as live or encrypted HLS, go straight to ffmpeg and start over. Partial files + stay until the job finishes or is removed, and finishing needs free space for + about twice the video's size. ## Adding a site diff --git a/internal/downloader/ffmpeg.go b/internal/downloader/ffmpeg.go index aec184d..d375409 100644 --- a/internal/downloader/ffmpeg.go +++ b/internal/downloader/ffmpeg.go @@ -4,10 +4,15 @@ import ( "bufio" "bytes" "context" + "errors" "fmt" "io" + "log/slog" "net/http" + "net/http/cookiejar" + "os" "os/exec" + "path/filepath" "strconv" "strings" "sync" @@ -21,28 +26,22 @@ import ( var defaultClient = &http.Client{Timeout: 30 * time.Second} -// FFmpeg remuxes streams into MP4 without transcoding. +const playlistTimeout = 30 * time.Second + +// FFmpeg downloads HLS segments itself, keeping them across attempts, and +// remuxes them into MP4 with ffmpeg. Other streams go straight to ffmpeg. +// Neither transcodes. type FFmpeg struct { - Path string // binary; "ffmpeg" if empty - Client *http.Client // for HLS master playlists; defaultClient if nil + Path string // binary; "ffmpeg" if empty + // Fetches playlists and segments; nil uses a default. Each download gets + // its own copy, with a cookie jar if the client has none. + Client *http.Client + Log *slog.Logger // nil discards // Zero defaults to 5 minutes, allowing for +faststart's silent final pass. StallTimeout time.Duration -} -// Select HLS inputs before invoking ffmpeg, which otherwise probes every variant. -func pickInputs(ctx context.Context, client *http.Client, s provider.Stream) ([]string, string, error) { - m, ok, err := hls.Load(ctx, client, s) - if err != nil { - return nil, "", err - } - if !ok { - return []string{s.URL}, "", nil - } - inputs := []string{m.Video.URI} - if m.Audio != "" { - inputs = append(inputs, m.Audio) - } - return inputs, m.AudioLanguage, nil + segmentTimeout time.Duration // zero defaults to a minute + retryDelay func(try int) time.Duration // nil doubles from a second } // ffmpeg logs these when it drops data but still exits 0. @@ -56,32 +55,90 @@ var dataLossMessages = []string{ "corrupt input packet in", } -func (f *FFmpeg) Download(parent context.Context, s provider.Stream, out string, progress func(time.Duration, int64)) error { - bin := f.Path - if bin == "" { - bin = "ffmpeg" +var ( + errDataLoss = errors.New("incomplete download") + errUnreadable = errors.New("unreadable input") +) + +// ffmpeg logs these for inputs it can't read at all; 5.x only the second. +var unreadableMessages = []string{ + "Error opening input", + "Invalid data found when processing input", +} + +// Segment requests share connections, one per worker. +var segmentTransport = func() *http.Transport { + t := http.DefaultTransport.(*http.Transport).Clone() + t.MaxIdleConnsPerHost = segmentWorkers + return t +}() + +// Download keeps segments in out's folder until a download succeeds or the +// stream changes. +func (f *FFmpeg) Download(ctx context.Context, s provider.Stream, out string, progress func(time.Duration, int64)) error { + client := &http.Client{Transport: segmentTransport} + if f.Client != nil { + c := *f.Client + client = &c } - client := f.Client - if client == nil { - client = defaultClient + if client.Jar == nil { + // ffmpeg sends a playlist's cookies with its segments too. + client.Jar, _ = cookiejar.New(nil) } - stall := f.StallTimeout - if stall == 0 { - stall = 5 * time.Minute + // Fail before fetching a stream that can't be remuxed. + if _, err := f.binary(); err != nil { + return err } - inputs, audioLanguage, err := pickInputs(parent, client, s) + // Select HLS inputs before invoking ffmpeg, which otherwise probes every variant. + loadCtx, cancel := context.WithTimeout(ctx, playlistTimeout) + m, master, err := hls.Load(loadCtx, client, s) + cancel() if err != nil { return err } - ctx, cancel := context.WithCancelCause(parent) - defer cancel(nil) - - args := []string{"-nostdin", "-hide_banner", "-loglevel", "warning", "-y"} + dir := filepath.Join(filepath.Dir(out), segmentsDir) + if master { + err := f.downloadSegments(ctx, client, s.Header, m, dir, out, progress) + if !errors.Is(err, hls.ErrUnsupported) { + return err + } + f.log().Info("can't fetch the stream's segments; ffmpeg downloads it from the start", + "file", filepath.Base(out), "err", err) + } + if err := os.RemoveAll(dir); err != nil { + return err + } + inputs := directInputs(s, m, master) + var args []string for _, in := range inputs { args = append(args, headerArgs(s.Header)...) args = append(args, "-i", in) } - if len(inputs) == 2 { + return f.run(ctx, "", append(args, outputArgs(len(inputs), m.AudioLanguage, out)...), progress) +} + +func (f *FFmpeg) log() *slog.Logger { + if f.Log == nil { + return slog.New(slog.DiscardHandler) + } + return f.Log +} + +func directInputs(s provider.Stream, m hls.Master, master bool) []string { + if !master { + return []string{s.URL} + } + if m.Audio != "" { + return []string{m.Video.URI, m.Audio} + } + return []string{m.Video.URI} +} + +// outputArgs follow inputs, whose first video and, if there are two, second's +// audio go into out. +func outputArgs(inputs int, audioLanguage, out string) []string { + var args []string + if inputs == 2 { args = append(args, "-map", "0:v:0", "-map", "1:a:0") } // Opening rendition URLs bypasses the master's language tag. MP4 needs a @@ -92,14 +149,48 @@ func (f *FFmpeg) Download(parent context.Context, s provider.Stream, out string, args = append(args, "-metadata:s:a:0", "language="+code) } } - args = append(args, + return append(args, "-sn", "-dn", // not all subtitle/data streams fit in MP4 "-c", "copy", "-movflags", "+faststart", "-progress", "pipe:1", "-nostats", out, ) +} + +// binary returns ffmpeg's absolute path: a relative one would resolve against +// the folder ffmpeg runs in. +func (f *FFmpeg) binary() (string, error) { + bin := f.Path + if bin == "" { + bin = "ffmpeg" + } + bin, err := exec.LookPath(bin) + if err == nil { + bin, err = filepath.Abs(bin) + } + if err != nil { + return "", fmt.Errorf("ffmpeg: %w", err) + } + return bin, nil +} + +// run runs ffmpeg in dir, or the current folder if dir is empty. +func (f *FFmpeg) run(parent context.Context, dir string, args []string, progress func(time.Duration, int64)) error { + bin, err := f.binary() + if err != nil { + return err + } + stall := f.StallTimeout + if stall == 0 { + stall = 5 * time.Minute + } + ctx, cancel := context.WithCancelCause(parent) + defer cancel(nil) + + args = append([]string{"-nostdin", "-hide_banner", "-loglevel", "warning", "-y"}, args...) cmd := exec.CommandContext(ctx, bin, args...) + cmd.Dir = dir cmd.WaitDelay = 5 * time.Second // Isolate ffmpeg from terminal signals so magnetowid can requeue interrupted jobs. cmd.SysProcAttr = ownProcessGroup() @@ -116,6 +207,7 @@ func (f *FFmpeg) Download(parent context.Context, s provider.Stream, out string, } tail := &tailBuffer{max: 4096} + var unreadable atomic.Bool stderrDone := make(chan struct{}) go func() { defer close(stderrDone) @@ -123,9 +215,14 @@ func (f *FFmpeg) Download(parent context.Context, s provider.Stream, out string, for sc.Scan() { line := sc.Text() tail.Write([]byte(line + "\n")) + for _, m := range unreadableMessages { + if strings.Contains(line, m) { + unreadable.Store(true) + } + } for _, m := range dataLossMessages { if strings.Contains(line, m) { - cancel(fmt.Errorf("incomplete download: %s", strings.TrimSpace(line))) + cancel(fmt.Errorf("%w: %s", errDataLoss, strings.TrimSpace(line))) } } } @@ -173,6 +270,9 @@ func (f *FFmpeg) Download(parent context.Context, s provider.Stream, out string, return fmt.Errorf("ffmpeg: %w", cause) } if waitErr != nil { + if unreadable.Load() { + waitErr = fmt.Errorf("%w: %w", errUnreadable, waitErr) + } return fmt.Errorf("ffmpeg: %w: %s", waitErr, strings.TrimSpace(tail.String())) } return nil diff --git a/internal/downloader/ffmpeg_test.go b/internal/downloader/ffmpeg_test.go index 92c0226..844fd73 100644 --- a/internal/downloader/ffmpeg_test.go +++ b/internal/downloader/ffmpeg_test.go @@ -15,6 +15,7 @@ import ( "testing" "time" + "github.com/combor/magnetowid/internal/hls" "github.com/combor/magnetowid/internal/provider" ) @@ -207,7 +208,6 @@ func segmentPath(t *testing.T, dir, name string, n int) string { return "/" + segs[n] } -// ffmpeg skips a missing segment and still exits 0. func TestFFmpegFailsOnMissingSegment(t *testing.T) { if _, err := exec.LookPath("ffmpeg"); err != nil { t.Skip("ffmpeg not installed") @@ -225,8 +225,14 @@ func TestFFmpegFailsOnMissingSegment(t *testing.T) { defer srv.Close() err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) + if err == nil || !strings.Contains(err.Error(), "video segment 2 of 3: HTTP 404") { + t.Errorf("segments: err = %v", err) + } + // ffmpeg skips a missing segment and still exits 0. + err = (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/stream_0.m3u8"}, + filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) if err == nil || !strings.Contains(err.Error(), "incomplete download") { - t.Fatalf("err = %v", err) + t.Errorf("ffmpeg: err = %v", err) } } @@ -246,8 +252,9 @@ func TestFFmpegStallTimeout(t *testing.T) { })) defer srv.Close() start := time.Now() + // ffmpeg reads media playlists itself. err := (&FFmpeg{StallTimeout: time.Second}).Download(context.Background(), - provider.Stream{URL: srv.URL + "/master.m3u8"}, filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) + provider.Stream{URL: srv.URL + "/stream_0.m3u8"}, filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) if err == nil || !strings.Contains(err.Error(), "stalled") { t.Fatalf("err = %v", err) } @@ -296,7 +303,7 @@ func TestFFmpegFailsOnCorruptTSSegment(t *testing.T) { } } -func TestPickInputs(t *testing.T) { +func TestDirectInputs(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/split.m3u8": @@ -327,9 +334,10 @@ video.m3u8 {"/media.m3u8", []string{srv.URL + "/media.m3u8"}, ""}, {"/film.mp4", []string{srv.URL + "/film.mp4"}, ""}, } { - got, language, err := pickInputs(ctx, srv.Client(), provider.Stream{URL: srv.URL + tt.path}) - if err != nil || !reflect.DeepEqual(got, tt.want) || language != tt.language { - t.Errorf("%s: got %v, %q, %v; want %v, %q", tt.path, got, language, err, tt.want, tt.language) + s := provider.Stream{URL: srv.URL + tt.path} + m, master, err := hls.Load(ctx, srv.Client(), s) + if got := directInputs(s, m, master); err != nil || !reflect.DeepEqual(got, tt.want) || m.AudioLanguage != tt.language { + t.Errorf("%s: got %v, %q, %v; want %v, %q", tt.path, got, m.AudioLanguage, err, tt.want, tt.language) } } } diff --git a/internal/downloader/ffmpeg_unix_test.go b/internal/downloader/ffmpeg_unix_test.go index 21e7a1d..3ba6bf8 100644 --- a/internal/downloader/ffmpeg_unix_test.go +++ b/internal/downloader/ffmpeg_unix_test.go @@ -4,8 +4,13 @@ package downloader import ( "context" + "errors" + "io" + "net/http" + "net/http/httptest" "os" "path/filepath" + "slices" "strconv" "strings" "syscall" @@ -52,3 +57,84 @@ func TestFFmpegRunsInOwnProcessGroup(t *testing.T) { t.Errorf("ffmpeg (pid %d) is in process group %d, not its own", pid, pgid) } } + +// fakeFFmpeg records its arguments and folder in log and creates its output. +func fakeFFmpeg(t *testing.T, dir string) (bin, log string) { + t.Helper() + bin, log = filepath.Join(dir, "ffmpeg"), filepath.Join(dir, "args") + script := "#!/bin/sh\npwd > '" + log + "'\nfor a; do echo \"$a\"; out=$a; done >> '" + log + "'\n: > \"$out\"\n" + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + return bin, log +} + +func TestFFmpegInputs(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + media := r.URL.RawQuery == "media" + switch { + case r.URL.Path == "/1.ts": + io.WriteString(w, "segment data") + case !media: + io.WriteString(w, "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\n"+strings.TrimPrefix(r.URL.Path, "/")+"?media\n") + case r.URL.Path == "/vod.m3u8": + io.WriteString(w, "#EXTM3U\n#EXTINF:4,\n1.ts\n#EXT-X-ENDLIST\n") + default: + io.WriteString(w, "#EXTM3U\n#EXTINF:4,\n1.ts\n") + } + })) + defer srv.Close() + dir := t.TempDir() + _, log := fakeFFmpeg(t, dir) + // A relative path must survive ffmpeg running in the segments folder. + t.Chdir(dir) + for _, tt := range []struct { + path, dir, input string + }{ + {"/vod.m3u8", filepath.Join(dir, segmentsDir), "v.m3u8"}, + {"/live.m3u8", dir, srv.URL + "/live.m3u8?media"}, + } { + err := (&FFmpeg{Path: "./ffmpeg"}).Download(context.Background(), provider.Stream{URL: srv.URL + tt.path}, + filepath.Join(dir, "out.mp4"), func(time.Duration, int64) {}) + if err != nil { + t.Fatalf("%s: %v", tt.path, err) + } + b, err := os.ReadFile(log) + if err != nil { + t.Fatal(err) + } + lines := strings.Split(string(b), "\n") + if lines[0] != tt.dir || !slices.Contains(lines, tt.input) { + t.Errorf("%s: ffmpeg ran in %s with %q; want %s and input %s", tt.path, lines[0], lines[1:], tt.dir, tt.input) + } + } +} + +// ffmpeg 5.x reports unreadable inputs without "Error opening input". +func TestUnreadableSegmentsRemoved(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/master.m3u8": + io.WriteString(w, "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\nvod.m3u8\n") + case "/vod.m3u8": + io.WriteString(w, "#EXTM3U\n#EXTINF:4,\n1.ts\n#EXT-X-ENDLIST\n") + default: + io.WriteString(w, "segment data") + } + })) + defer srv.Close() + dir := t.TempDir() + bin := filepath.Join(dir, "ffmpeg") + script := "#!/bin/sh\necho 'v.m3u8: Invalid data found when processing input' >&2\nexit 1\n" + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + err := (&FFmpeg{Path: bin}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, + filepath.Join(dir, "out.mp4"), func(time.Duration, int64) {}) + if !errors.Is(err, errUnreadable) { + t.Errorf("err = %v", err) + } + if _, err := os.Stat(filepath.Join(dir, segmentsDir)); !os.IsNotExist(err) { + t.Errorf("unreadable segments kept: %v", err) + } +} diff --git a/internal/downloader/queue.go b/internal/downloader/queue.go index 4fe093b..0f9bc0b 100644 --- a/internal/downloader/queue.go +++ b/internal/downloader/queue.go @@ -57,6 +57,8 @@ var ( errPaused = errors.New("paused") ) +// Engine may keep partial downloads in out's folder, which stays until the job +// finishes or is removed. type Engine interface { Download(ctx context.Context, s provider.Stream, out string, progress func(done time.Duration, bytes int64)) error } @@ -81,6 +83,8 @@ type Job struct { Priority int `json:"priority,omitzero"` // -1 low, 0 normal, 1 high, 2 force // Separate from Status so older versions can still load paused jobs. Paused bool `json:"paused,omitzero"` + + startFraction float64 // when this attempt's progress began; not saved } // Assume 4 Mbit/s until ffmpeg reports progress. @@ -96,10 +100,11 @@ func (j Job) EstimatedSize() int64 { // TimeLeft is zero until a running download has made some progress. func (j Job) TimeLeft(now time.Time) time.Duration { - if j.Status != StatusDownloading || j.Fraction <= 0.01 { + gained := j.Fraction - j.startFraction + if j.Status != StatusDownloading || gained <= 0.01 { return 0 } - return time.Duration(float64(now.Sub(j.Started)) * (1 - j.Fraction) / j.Fraction) + return time.Duration(float64(now.Sub(j.Started)) * (1 - j.Fraction) / gained) } // HistoryRetention is how long finished jobs are kept. @@ -141,10 +146,6 @@ func New(dir string, db *bolt.DB, providers *provider.Registry, engine Engine, l if err != nil { return nil, err } - // Before the worker starts, all work folders are crash leftovers. - if err := os.RemoveAll(filepath.Join(dir, incompleteDir)); err != nil { - log.Warn("removing unfinished downloads", "err", err) - } q := &Queue{ dir: dir, providers: providers, @@ -163,9 +164,31 @@ func New(dir string, db *bolt.DB, providers *provider.Registry, engine Engine, l q.order = append(q.order, j.ID) } q.prune(time.Now()) + q.cleanWork() return q, nil } +// cleanWork removes partial downloads that no job will continue. +func (q *Queue) cleanWork() { + entries, err := os.ReadDir(filepath.Join(q.dir, incompleteDir)) + if err != nil && !errors.Is(err, fs.ErrNotExist) { + q.log.Warn("listing unfinished downloads", "err", err) + } + for _, e := range entries { + if j := q.jobs[e.Name()]; j == nil || j.Status == StatusCompleted || j.Status == StatusFailed { + q.removeWork(e.Name()) + } + } +} + +func (q *Queue) workDir(id string) string { return filepath.Join(q.dir, incompleteDir, id) } + +func (q *Queue) removeWork(id string) { + if err := os.RemoveAll(q.workDir(id)); err != nil { + q.log.Warn("removing an unfinished download", "id", id, "err", err) + } +} + func (q *Queue) Dir() string { return q.dir } // Add returns a SABnzbd job ID only after saving the job. Higher priorities run first. @@ -220,8 +243,8 @@ func (q *Queue) Outages() map[string]time.Time { return out } -// SetPaused stops or resumes the queue. Resumed downloads restart from the beginning. -// A save failure leaves the queue unchanged. +// SetPaused stops or resumes the queue. Stopped downloads continue where they +// left off. A save failure leaves the queue unchanged. func (q *Queue) SetPaused(paused bool) error { q.mu.Lock() defer q.mu.Unlock() @@ -244,8 +267,8 @@ func (q *Queue) SetPaused(paused bool) error { return nil } -// PauseJobs returns affected IDs. Resumed downloads restart from the beginning. -// On a save failure, earlier changes remain applied. +// PauseJobs returns affected IDs. Stopped downloads continue where they left +// off. On a save failure, earlier changes remain applied. func (q *Queue) PauseJobs(ids ...string) ([]string, error) { return q.setJobsPaused(true, ids) } @@ -329,6 +352,8 @@ func (q *Queue) discard(id string, deleteFiles, unfinishedOnly bool) (bool, erro q.mu.Unlock() return false, err } + // A running job's worker removes its files once the engine stops. + running := job.Status == StatusDownloading delete(q.jobs, id) for i, v := range q.order { if v == id { @@ -348,6 +373,9 @@ func (q *Queue) discard(id string, deleteFiles, unfinishedOnly bool) (bool, erro if cancel != nil { cancel(nil) } + if !running { + q.removeWork(id) + } q.log.Info("job deleted", "id", id, "delete_files", deleteFiles) return true, nil } @@ -414,6 +442,7 @@ func (q *Queue) next() (id string, wait time.Duration, ok bool) { } best.Status = StatusDownloading best.Started = now + best.startFraction = best.Fraction best.Attempts++ return best.ID, 0, true } @@ -426,6 +455,8 @@ func (q *Queue) process(parent context.Context, id string) { job, ok := q.jobs[id] if !ok { q.mu.Unlock() + // Deleted after next selected it, so discard left the files to us. + q.removeWork(id) return } q.cancels[id] = cancel @@ -437,22 +468,26 @@ func (q *Queue) process(parent context.Context, id string) { q.mu.Unlock() q.log.Info("job started", "id", id, "name", j.Name) - work := filepath.Join(q.dir, incompleteDir, id) - storage, err := q.run(ctx, id, j, work) - if rmErr := os.RemoveAll(work); rmErr != nil { - q.log.Warn("removing work dir", "id", id, "err", rmErr) + storage, err := q.run(ctx, id, j, q.workDir(id)) + // Finished and deleted jobs never run again, so their files can go unlocked. + if !q.finish(parent, ctx, id, storage, err) { + q.removeWork(id) } +} +// finish records an attempt's outcome. keep reports whether the job may +// continue its partial download. +func (q *Queue) finish(parent, ctx context.Context, id, storage string, err error) (keep bool) { q.mu.Lock() defer q.mu.Unlock() delete(q.cancels, id) - job, ok = q.jobs[id] + job, ok := q.jobs[id] if !ok { // Deleted while running. if storage != "" { os.RemoveAll(storage) } - return + return false } // Persist outcomes, not running progress, so crashes leave jobs queued for retry. defer func() { @@ -465,12 +500,12 @@ func (q *Queue) process(parent context.Context, id string) { // Shutdown does not consume a retry. requeue(job) q.log.Info("job interrupted, requeued", "id", id, "name", job.Name) - return + return true } if err != nil && errors.Is(context.Cause(ctx), errPaused) { requeue(job) q.log.Info("job stopped by a pause, requeued", "id", id, "name", job.Name) - return + return true } if errors.Is(err, errUnreachable) { // Provider outages pause all its jobs without consuming retries. @@ -486,7 +521,7 @@ func (q *Queue) process(parent context.Context, id string) { o.until = time.Now().Add(delay) q.log.Warn("pausing provider's jobs", "provider", job.Ref.Provider, "retry_in", delay, "err", err) - return + return true } delete(q.outages, job.Ref.Provider) job.Finished = time.Now() @@ -496,20 +531,20 @@ func (q *Queue) process(parent context.Context, id string) { delay := q.retryDelay(job.Attempts) job.Status = StatusQueued job.RetryAt = job.Finished.Add(delay) - job.Fraction, job.Bytes = 0, 0 q.log.Warn("job attempt failed, will retry", "id", id, "name", job.Name, "attempt", job.Attempts, "retry_in", delay, "err", err) - return + return true } job.Status = StatusFailed q.log.Error("job failed", "id", id, "name", job.Name, "attempts", job.Attempts, "err", err) - return + return false } job.Status = StatusCompleted job.Error = "" job.Fraction = 1 job.Storage = storage q.log.Info("job completed", "id", id, "name", job.Name, "storage", storage, "bytes", job.Bytes) + return false } // prune removes expired history but keeps downloaded files. Requires q.mu. @@ -536,11 +571,10 @@ func (q *Queue) prune(now time.Time) { q.log.Info("forgot old finished jobs", "count", len(old)) } -// Requeue without consuming a retry. +// Requeue without consuming a retry. Progress stays with the partial download. func requeue(job *Job) { job.Status = StatusQueued job.Attempts-- - job.Fraction, job.Bytes = 0, 0 } // offline distinguishes network failures from item-specific errors. @@ -568,6 +602,7 @@ func (q *Queue) run(ctx context.Context, id string, j Job, work string) (string, } out := filepath.Join(work, j.Name+".mp4") duration := time.Duration(j.Ref.Duration) * time.Second + first := true progress := func(done time.Duration, bytes int64) { q.mu.Lock() defer q.mu.Unlock() @@ -579,6 +614,10 @@ func (q *Queue) run(ctx context.Context, id string, j Job, work string) (string, if duration > 0 { job.Fraction = min(float64(done)/float64(duration), 0.99) } + // The first report includes what earlier attempts saved. + if first { + job.startFraction, first = job.Fraction, false + } } // Stream URLs expire, so resolve on every attempt. diff --git a/internal/downloader/queue_test.go b/internal/downloader/queue_test.go index 7fa4f61..fc63a1c 100644 --- a/internal/downloader/queue_test.go +++ b/internal/downloader/queue_test.go @@ -188,9 +188,7 @@ func TestJobCompletes(t *testing.T) { if j.Fraction != 1 || j.Bytes != 7 { t.Errorf("fraction = %v, bytes = %d", j.Fraction, j.Bytes) } - if _, err := os.Stat(filepath.Join(q.Dir(), ".incomplete", id)); !os.IsNotExist(err) { - t.Errorf("work dir not removed: %v", err) - } + waitGone(t, q.workDir(id)) } func TestSameNameGetsSuffix(t *testing.T) { @@ -597,18 +595,212 @@ func TestCrashedJobIsQueued(t *testing.T) { } } -func TestNewRemovesLeftoverWork(t *testing.T) { +func TestNewKeepsOnlyUnfinishedWork(t *testing.T) { dir := t.TempDir() - stale := filepath.Join(dir, incompleteDir, "SABnzbd_nzo_x", "x.mp4") - if err := os.MkdirAll(filepath.Dir(stale), 0o777); err != nil { - t.Fatal(err) + q := newQueue(t, dir, &fakeProvider{}, &fakeEngine{}) + queued := add(t, q, "queued", "tv", 0, nzb.Ref{Provider: "fake", ID: "queued"}) + done := add(t, q, "done", "tv", 0, nzb.Ref{Provider: "fake", ID: "done"}) + q.mu.Lock() + q.jobs[done].Status = StatusCompleted + q.put(q.jobs[done]) + q.mu.Unlock() + q.db.Close() + for _, id := range []string{queued, done, "SABnzbd_nzo_gone"} { + if err := os.MkdirAll(filepath.Join(dir, incompleteDir, id), 0o777); err != nil { + t.Fatal(err) + } } - if err := os.WriteFile(stale, []byte("partial"), 0o666); err != nil { + if err := os.WriteFile(filepath.Join(dir, incompleteDir, "stray"), nil, 0o666); err != nil { t.Fatal(err) } + newQueue(t, dir, &fakeProvider{}, &fakeEngine{}) - if _, err := os.Stat(filepath.Join(dir, incompleteDir)); !os.IsNotExist(err) { - t.Errorf("leftover work not removed: %v", err) + entries, err := os.ReadDir(filepath.Join(dir, incompleteDir)) + if err != nil || len(entries) != 1 || entries[0].Name() != queued { + t.Errorf("kept %v, %v; want only %s", entries, err, queued) + } +} + +// workEngine leaves a partial download and tells found whether an earlier +// attempt left one. Its first fail calls fail; then it blocks until cancelled +// if block is set, or else completes. +type workEngine struct { + found chan bool + block bool + fail int + err error + + mu sync.Mutex + calls int +} + +func newWorkEngine() *workEngine { return &workEngine{found: make(chan bool, 10)} } + +func (e *workEngine) Download(ctx context.Context, _ provider.Stream, out string, progress func(time.Duration, int64)) error { + e.mu.Lock() + e.calls++ + failing := e.calls <= e.fail + e.mu.Unlock() + partial := filepath.Join(filepath.Dir(out), "partial") + _, err := os.Stat(partial) + e.found <- err == nil + if err := os.WriteFile(partial, nil, 0o666); err != nil { + return err + } + progress(30*time.Second, 7) + switch { + case failing: + return e.err + case e.block: + <-ctx.Done() + return ctx.Err() + } + return os.WriteFile(out, []byte("media"), 0o644) +} + +func waitGone(t *testing.T, path string) { + t.Helper() + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { + if _, err := os.Stat(path); os.IsNotExist(err) { + return + } + } + t.Fatalf("%s not removed", path) +} + +func exists(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); err != nil { + t.Fatal(err) + } +} + +func TestPauseKeepsWork(t *testing.T) { + e := newWorkEngine() + e.block = true + q := newQueue(t, t.TempDir(), &fakeProvider{}, e) + run(t, q) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a", Duration: 60}) + if <-e.found { + t.Fatal("a new job found a partial download") + } + if _, err := q.PauseJobs(id); err != nil { + t.Fatal(err) + } + j := waitJob(t, q, id, func(j Job) bool { return j.Status == StatusQueued }) + if j.Fraction != 0.5 || j.Bytes != 7 { + t.Errorf("paused at %v, %d bytes; want the progress kept", j.Fraction, j.Bytes) + } + exists(t, filepath.Join(q.workDir(id), "partial")) + if _, err := q.ResumeJobs(id); err != nil { + t.Fatal(err) + } + if !<-e.found { + t.Error("the resumed job lost its partial download") + } +} + +func TestShutdownKeepsWork(t *testing.T) { + dir := t.TempDir() + e := newWorkEngine() + e.block = true + q := newQueue(t, dir, &fakeProvider{}, e) + stop := run(t, q) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a"}) + <-e.found + stop() + q.db.Close() + + e = newWorkEngine() + q = newQueue(t, dir, &fakeProvider{}, e) + run(t, q) + if !<-e.found { + t.Error("the restarted job lost its partial download") + } + waitFinished(t, q, id) + waitGone(t, q.workDir(id)) +} + +func TestRetryKeepsWork(t *testing.T) { + e := newWorkEngine() + e.fail, e.err = 1, errors.New("boom") + q := startQueue(t, &fakeProvider{}, e) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a"}) + if <-e.found || !<-e.found { + t.Error("the retry lost the partial download") + } + if j := waitFinished(t, q, id); j.Status != StatusCompleted { + t.Fatalf("status %s", j.Status) + } + waitGone(t, q.workDir(id)) +} + +func TestFailureRemovesWork(t *testing.T) { + e := newWorkEngine() + e.fail, e.err = 1, fmt.Errorf("%w: DRM", provider.ErrUnavailable) + q := startQueue(t, &fakeProvider{}, e) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a"}) + if j := waitFinished(t, q, id); j.Status != StatusFailed { + t.Fatalf("status %s", j.Status) + } + waitGone(t, q.workDir(id)) +} + +func TestRemovingJobRemovesWork(t *testing.T) { + t.Run("running", func(t *testing.T) { + e := newWorkEngine() + e.block = true + q := startQueue(t, &fakeProvider{}, e) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a"}) + <-e.found + if ok, err := q.Delete(id, false); !ok || err != nil { + t.Fatalf("Delete = %v, %v", ok, err) + } + waitGone(t, q.workDir(id)) + }) + t.Run("paused", func(t *testing.T) { + e := newWorkEngine() + e.block = true + q := startQueue(t, &fakeProvider{}, e) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a"}) + <-e.found + if _, err := q.PauseJobs(id); err != nil { + t.Fatal(err) + } + waitJob(t, q, id, func(j Job) bool { return j.Status == StatusQueued }) + exists(t, q.workDir(id)) + if ok, err := q.Cancel(id); !ok || err != nil { + t.Fatalf("Cancel = %v, %v", ok, err) + } + if _, err := os.Stat(q.workDir(id)); !os.IsNotExist(err) { + t.Errorf("partial download kept: %v", err) + } + }) + t.Run("waiting to retry", func(t *testing.T) { + e := newWorkEngine() + e.fail, e.err = 1, errors.New("boom") + q := startQueueWithDelay(t, &fakeProvider{}, e, time.Hour) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "a"}) + waitJob(t, q, id, func(j Job) bool { return j.Status == StatusQueued && j.Attempts == 1 }) + exists(t, q.workDir(id)) + if ok, err := q.Delete(id, false); !ok || err != nil { + t.Fatalf("Delete = %v, %v", ok, err) + } + if _, err := os.Stat(q.workDir(id)); !os.IsNotExist(err) { + t.Errorf("partial download kept: %v", err) + } + }) +} + +func TestTimeLeftAfterResume(t *testing.T) { + now := time.Now() + j := Job{Status: StatusDownloading, Started: now.Add(-10 * time.Second), Fraction: 0.6, startFraction: 0.5} + if got := j.TimeLeft(now); got != 40*time.Second { + t.Errorf("TimeLeft = %v, want 40s", got) + } + j.Fraction = 0.505 + if got := j.TimeLeft(now); got != 0 { + t.Errorf("TimeLeft without progress = %v, want 0", got) } } diff --git a/internal/downloader/segments.go b/internal/downloader/segments.go new file mode 100644 index 0000000..58d709c --- /dev/null +++ b/internal/downloader/segments.go @@ -0,0 +1,447 @@ +package downloader + +import ( + "cmp" + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "math" + "mime" + "net/http" + "net/url" + "os" + "path" + "path/filepath" + "slices" + "strings" + "sync" + "time" + + "github.com/combor/magnetowid/internal/hls" +) + +const ( + segmentsDir = "segments" + segmentWorkers = 4 + segmentTries = 5 + // Bump to discard segments saved in an older layout. + segmentsLayout = 1 +) + +var errStalled = errors.New("stalled") + +// track is a rendition saved in its own folder, listed in a local playlist of +// the same name. +type track struct { + name, kind string // "v" and "video", or "a" and "audio" + uri string // media playlist + source string // identifies the rendition + segments []hls.Segment + inits []string // distinct initialization sections +} + +// part is a file to fetch. +type part struct { + uri string + path string // slash-separated, relative to the segments folder + start time.Duration // playback order + video time.Duration // playback time a video segment adds + label string // e.g. "video segment 12 of 900" +} + +// downloadSegments saves the stream's segments in dir, keeping those an +// earlier attempt saved, and remuxes them into out. It returns +// hls.ErrUnsupported before touching dir. +func (f *FFmpeg) downloadSegments(ctx context.Context, client *http.Client, h http.Header, m hls.Master, + dir, out string, progress func(time.Duration, int64)) error { + tracks, err := loadTracks(ctx, client, h, m) + if err != nil { + return err + } + stale, err := prepare(dir, tracks) + if err != nil { + return err + } + if stale { + f.log().Info("saved segments are of another stream; starting over", "file", filepath.Base(out)) + } + parts := layout(tracks) + pending, done, bytes := scan(dir, parts) + progress(done, bytes) + if saved := len(parts) - len(pending); saved > 0 { + f.log().Info("resuming download", "file", filepath.Base(out), "saved", saved, "files", len(parts)) + } + var mu sync.Mutex + err = f.fetchAll(ctx, client, h, dir, pending, func(p part, n int64) { + mu.Lock() + defer mu.Unlock() + done += p.video + bytes += n + progress(done, bytes) + }) + if err != nil { + return err + } + + out, err = filepath.Abs(out) + if err != nil { + return err + } + args := []string{"-i", tracks[0].name + ".m3u8"} + if len(tracks) == 2 { + args = append(args, "-i", tracks[1].name+".m3u8") + } + err = f.run(ctx, dir, append(args, outputArgs(len(tracks), m.AudioLanguage, out)...), func(time.Duration, int64) {}) + if errors.Is(err, errDataLoss) || errors.Is(err, errUnreadable) { + // A damaged file on disk would fail every retry. + if rmErr := os.RemoveAll(dir); rmErr != nil { + f.log().Warn("removing damaged segments", "file", filepath.Base(out), "err", rmErr) + } + } + if err != nil { + return err + } + fi, err := os.Stat(out) + if err != nil { + return err + } + progress(done, fi.Size()) + return nil +} + +func loadTracks(ctx context.Context, client *http.Client, h http.Header, m hls.Master) ([]track, error) { + ctx, cancel := context.WithTimeout(ctx, playlistTimeout) + defer cancel() + tracks := []track{{name: "v", kind: "video", uri: m.Video.URI, source: fmt.Sprintf("%dx%d %d %s %s", + m.Video.Width, m.Video.Height, m.Video.Bandwidth, m.Video.Codecs, baseName(m.Video.URI))}} + if m.Audio != "" { + tracks = append(tracks, track{name: "a", kind: "audio", uri: m.Audio, + source: fmt.Sprintf("%s %s", m.AudioLanguage, baseName(m.Audio))}) + } + for i := range tracks { + t := &tracks[i] + segs, err := hls.LoadMedia(ctx, client, t.uri, h) + if err != nil { + return nil, err + } + t.segments = segs + // Files are saved by position, so their names must identify the stream too. + names := sha256.New() + for _, s := range segs { + if s.Init != "" && !slices.Contains(t.inits, s.Init) { + t.inits = append(t.inits, s.Init) + fmt.Fprintln(names, baseName(s.Init)) + } + fmt.Fprintln(names, baseName(s.URI)) + } + t.source += fmt.Sprintf(" files %x", names.Sum(nil)[:8]) + } + return tracks, nil +} + +// baseName leaves out the query, where tokens change with every resolve. +func baseName(uri string) string { + u, err := url.Parse(uri) + if err != nil { + return "" + } + return path.Base(u.Path) +} + +// prepare keeps saved files only if the tracks' local playlists are unchanged; +// stale reports discarded ones. +func prepare(dir string, tracks []track) (stale bool, err error) { + want := map[string]string{"v": "", "a": ""} + for _, t := range tracks { + want[t.name] = t.playlist() + } + same := true + for name, w := range want { + got, err := os.ReadFile(filepath.Join(dir, name+".m3u8")) + stale = stale || err == nil + same = same && string(got) == w + } + if same { + return false, nil + } + if err := os.RemoveAll(dir); err != nil { + return false, err + } + for _, t := range tracks { + if err := os.MkdirAll(filepath.Join(dir, t.name), 0o777); err != nil { + return false, err + } + } + for _, t := range tracks { + if err := os.WriteFile(filepath.Join(dir, t.name+".m3u8"), []byte(t.playlist()), 0o666); err != nil { + return false, err + } + } + return stale, nil +} + +// playlist lists the track's local files for ffmpeg. Its first lines identify +// the rendition, so the playlist changes with the stream. +func (t track) playlist() string { + var target float64 + for _, s := range t.segments { + target = max(target, math.Ceil(s.Duration.Seconds())) + } + var b strings.Builder + fmt.Fprintf(&b, "#EXTM3U\n## magnetowid layout %d: %s\n#EXT-X-TARGETDURATION:%d\n#EXT-X-PLAYLIST-TYPE:VOD\n", + segmentsLayout, t.source, int(target)) + init := "" + for i, s := range t.segments { + if s.Discontinuity { + b.WriteString("#EXT-X-DISCONTINUITY\n") + } + if s.Init != init { + init = s.Init + fmt.Fprintf(&b, "#EXT-X-MAP:URI=%q\n", t.initPath(init)) + } + fmt.Fprintf(&b, "#EXTINF:%.6f,\n%s\n", s.Duration.Seconds(), t.segmentPath(i)) + } + b.WriteString("#EXT-X-ENDLIST\n") + return b.String() +} + +func (t track) initPath(uri string) string { + return fmt.Sprintf("%s/init%d.mp4", t.name, slices.Index(t.inits, uri)) +} + +func (t track) segmentPath(i int) string { + return fmt.Sprintf("%s/%06d%s", t.name, i, segmentExt(t.segments[i])) +} + +// Use extensions that ffmpeg accepts for local segments of each format. +func segmentExt(s hls.Segment) string { + if s.Init != "" { + return ".mp4" + } + switch ext := strings.ToLower(path.Ext(baseName(s.URI))); ext { + case ".aac", ".ac3", ".eac3", ".mp3": + return ext + case ".ec3": + return ".eac3" + case ".m4s", ".m4a", ".m4v", ".mp4", ".cmfa", ".cmfv", ".fmp4": + return ".mp4" + } + return ".ts" +} + +// layout lists initialization sections, then segments in playback order so +// that video and audio progress together. +func layout(tracks []track) []part { + var inits, segs []part + for _, t := range tracks { + for i, uri := range t.inits { + inits = append(inits, part{uri: uri, path: t.initPath(uri), + label: fmt.Sprintf("%s initialization section %d", t.kind, i+1)}) + } + var start time.Duration + for i, s := range t.segments { + p := part{uri: s.URI, path: t.segmentPath(i), start: start, + label: fmt.Sprintf("%s segment %d of %d", t.kind, i+1, len(t.segments))} + if t.name == "v" { + p.video = s.Duration + } + segs = append(segs, p) + start += s.Duration + } + } + slices.SortStableFunc(segs, func(a, b part) int { return cmp.Compare(a.start, b.start) }) + return append(inits, segs...) +} + +// scan returns the parts still to fetch, and the video time and bytes saved. +func scan(dir string, parts []part) (pending []part, done time.Duration, bytes int64) { + for _, p := range parts { + // Files appear only when complete, and never empty. + fi, err := os.Stat(filepath.Join(dir, filepath.FromSlash(p.path))) + if err == nil && fi.Size() > 0 { + done += p.video + bytes += fi.Size() + continue + } + pending = append(pending, p) + } + return pending, done, bytes +} + +// fetchAll stops at the first error. It returns only after every worker has. +func (f *FFmpeg) fetchAll(parent context.Context, client *http.Client, h http.Header, dir string, + parts []part, saved func(part, int64)) error { + ctx, cancel := context.WithCancelCause(parent) + defer cancel(nil) + todo := make(chan part) + var wg sync.WaitGroup + for range segmentWorkers { + wg.Go(func() { + for p := range todo { + n, err := f.fetch(ctx, client, h, dir, p) + if err != nil { + cancel(err) + return + } + saved(p, n) + } + }) + } +feed: + for _, p := range parts { + select { + case todo <- p: + case <-ctx.Done(): + break feed + } + } + close(todo) + wg.Wait() + if err := parent.Err(); err != nil { + return err + } + return context.Cause(ctx) +} + +// fetch tries again after transient errors. +func (f *FFmpeg) fetch(ctx context.Context, client *http.Client, h http.Header, dir string, p part) (int64, error) { + for try := 1; ; try++ { + n, err := f.fetchOnce(ctx, client, h, dir, p) + if err == nil { + return n, nil + } + if ctx.Err() != nil { + return 0, ctx.Err() + } + err = fmt.Errorf("%s: %w", p.label, err) + if try == segmentTries || !transient(err) { + return 0, err + } + delay := time.Second << (try - 1) + if f.retryDelay != nil { + delay = f.retryDelay(try) + } + f.log().Debug("fetching again", "try", try, "in", delay, "err", err) + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-time.After(delay): + } + } +} + +func (f *FFmpeg) fetchOnce(ctx context.Context, client *http.Client, h http.Header, dir string, p part) (n int64, err error) { + timeout := cmp.Or(f.segmentTimeout, time.Minute) + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + idle := time.AfterFunc(timeout, func() { cancel(fmt.Errorf("%w: no data for %v", errStalled, timeout)) }) + defer idle.Stop() + defer func() { + if cause := context.Cause(ctx); err != nil && errors.Is(cause, errStalled) { + err = cause + } + }() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.uri, nil) + if err != nil { + return 0, permanent{err} + } + for k, vs := range h { + for _, v := range vs { + req.Header.Add(k, v) + } + } + resp, err := client.Do(req) + if err != nil { + // Leave tokenized URLs out of job errors. + var ue *url.Error + if errors.As(err, &ue) { + err = ue.Err + } + return 0, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return 0, &statusError{resp.StatusCode} + } + // ffmpeg would skip an error page saved as a segment without a word. + if mt, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")); mt == "text/html" { + return 0, errors.New("a web page instead of media") + } + + dst := filepath.Join(dir, filepath.FromSlash(p.path)) + tmp := dst + ".part" + file, err := os.Create(tmp) + if err != nil { + return 0, permanent{err} + } + n, err = io.Copy(diskWriter{file}, idleReader{resp.Body, idle, timeout}) + if err == nil && n == 0 { + err = errors.New("empty response") + } + if err == nil { + err = diskError(file.Sync()) + } + if closeErr := file.Close(); err == nil { + err = diskError(closeErr) + } + if err == nil { + err = diskError(os.Rename(tmp, dst)) + } + if err != nil { + os.Remove(tmp) + return 0, err + } + return n, nil +} + +// idleReader restarts the stall timer whenever data arrives. +type idleReader struct { + r io.Reader + timer *time.Timer + d time.Duration +} + +func (r idleReader) Read(p []byte) (int, error) { + n, err := r.r.Read(p) + if n > 0 { + r.timer.Reset(r.d) + } + return n, err +} + +// diskWriter tells write errors from read errors in io.Copy. +type diskWriter struct{ w io.Writer } + +func (d diskWriter) Write(p []byte) (int, error) { + n, err := d.w.Write(p) + return n, diskError(err) +} + +func diskError(err error) error { + if err != nil { + return permanent{err} + } + return nil +} + +// permanent errors recur when fetching again. +type permanent struct{ error } + +func (p permanent) Unwrap() error { return p.error } + +type statusError struct{ code int } + +func (e *statusError) Error() string { return fmt.Sprintf("HTTP %d", e.code) } + +// transient errors are server errors, rate limits, and lost or stalled connections. +func transient(err error) bool { + var se *statusError + if errors.As(err, &se) { + return se.code >= 500 || se.code == http.StatusRequestTimeout || se.code == http.StatusTooManyRequests + } + var p permanent + return !errors.As(err, &p) +} diff --git a/internal/downloader/segments_test.go b/internal/downloader/segments_test.go new file mode 100644 index 0000000..08b8387 --- /dev/null +++ b/internal/downloader/segments_test.go @@ -0,0 +1,515 @@ +package downloader + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/combor/magnetowid/internal/hls" + "github.com/combor/magnetowid/internal/provider" +) + +func needFFmpeg(t *testing.T) { + t.Helper() + if _, err := exec.LookPath("ffmpeg"); err != nil { + t.Skip("ffmpeg not installed") + } +} + +func noRetryDelay(int) time.Duration { return 0 } + +// requests counts requests by path. +type requests struct { + mu sync.Mutex + n map[string]int +} + +func (r *requests) add(path string) { + r.mu.Lock() + defer r.mu.Unlock() + if r.n == nil { + r.n = map[string]int{} + } + r.n[path]++ +} + +func (r *requests) get(path string) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.n[path] +} + +func savedFiles(t *testing.T, dir string) []string { + t.Helper() + var files []string + err := filepath.WalkDir(dir, func(p string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() { + files = append(files, p) + } + return err + }) + if err != nil { + t.Fatal(err) + } + return files +} + +func TestSegmentsResumeAfterCancel(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + blocked := segmentPath(t, src, "stream_0.m3u8", 2) + var block atomic.Bool + block.Store(true) + reached := make(chan struct{}, 1) + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reqs.add(r.URL.Path) + if r.URL.Path == blocked && block.Load() { + reached <- struct{}{} + <-r.Context().Done() + return + } + http.FileServer(http.Dir(src)).ServeHTTP(w, r) + })) + defer srv.Close() + work := t.TempDir() + out := filepath.Join(work, "out.mp4") + s := provider.Stream{URL: srv.URL + "/master.m3u8"} + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- (&FFmpeg{}).Download(ctx, s, out, func(time.Duration, int64) {}) }() + <-reached + // Let the other workers finish what they hold. + time.Sleep(200 * time.Millisecond) + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled download: err = %v", err) + } + var saved []string + for _, f := range savedFiles(t, filepath.Join(work, segmentsDir)) { + if strings.HasSuffix(f, ".part") { + t.Errorf("partial file left: %s", f) + } + if !strings.HasSuffix(f, ".m3u8") { + saved = append(saved, f) + } + } + if len(saved) < 3 { + t.Fatalf("saved %v, want at least the initialization sections and the first video segments", saved) + } + + block.Store(false) + before := map[string]int{} + for _, p := range []string{"/init_0.mp4", segmentPath(t, src, "stream_0.m3u8", 0), segmentPath(t, src, "stream_0.m3u8", 1)} { + before[p] = reqs.get(p) + } + var first []int64 + err := (&FFmpeg{}).Download(context.Background(), s, out, func(done time.Duration, bytes int64) { + if first == nil { + first = []int64{int64(done), bytes} + } + }) + if err != nil { + t.Fatal(err) + } + for p, n := range before { + if got := reqs.get(p); got != n { + t.Errorf("%s fetched again", p) + } + } + if first == nil || first[0] < int64(2*time.Second) || first[1] == 0 { + t.Errorf("first progress (done, bytes) = %v, want the saved segments", first) + } + probe, err := exec.Command("ffprobe", "-v", "error", "-show_entries", "stream=codec_type", + "-of", "csv=p=0", out).Output() + if err != nil { + t.Fatal(err) + } + if got := strings.Fields(string(probe)); len(got) != 2 { + t.Errorf("streams = %v, want video and audio", got) + } + checkAudioLanguage(t, out, "pol") +} + +func TestSegmentsRestartForAnotherRendition(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + master, err := os.ReadFile(filepath.Join(src, "master.m3u8")) + if err != nil { + t.Fatal(err) + } + // Offer only the low variant from now on. + low := strings.Replace(string(master), "stream_0.m3u8", "stream_1.m3u8", 1) + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reqs.add(r.URL.Path) + if r.URL.Path == "/low.m3u8" { + w.Write([]byte(low)) + return + } + http.FileServer(http.Dir(src)).ServeHTTP(w, r) + })) + defer srv.Close() + work := t.TempDir() + out := filepath.Join(work, "out.mp4") + noop := func(time.Duration, int64) {} + if err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, out, noop); err != nil { + t.Fatal(err) + } + if err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/low.m3u8"}, out, noop); err != nil { + t.Fatal(err) + } + if n := reqs.get("/init_2.mp4"); n != 2 { + t.Errorf("audio fetched %d times, want again for the new video", n) + } + probe, err := exec.Command("ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width", + "-of", "csv=p=0", out).Output() + if err != nil || strings.TrimSpace(string(probe)) != "80" { + t.Errorf("width = %q, %v; want the low variant's 80", probe, err) + } +} + +func TestSegmentsRetry(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + seg := segmentPath(t, src, "stream_0.m3u8", 1) + for _, tt := range []struct { + name string + statuses []int // then 200 + requests int + err string + }{ + {"not found", []int{404}, 1, "video segment 2 of 3: HTTP 404"}, + {"server errors", []int{503, 503, 503, 503, 503, 503}, segmentTries, "video segment 2 of 3: HTTP 503"}, + {"one server error", []int{500}, 2, ""}, + {"rate limited", []int{429}, 2, ""}, + {"error page", []int{200}, 2, ""}, + } { + t.Run(tt.name, func(t *testing.T) { + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == seg { + reqs.add(seg) + if n := reqs.get(seg); n <= len(tt.statuses) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.WriteHeader(tt.statuses[n-1]) + w.Write([]byte("Try again later")) + return + } + } + http.FileServer(http.Dir(src)).ServeHTTP(w, r) + })) + defer srv.Close() + err := (&FFmpeg{retryDelay: noRetryDelay}).Download(context.Background(), + provider.Stream{URL: srv.URL + "/master.m3u8"}, filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) + if tt.err == "" && err != nil || tt.err != "" && (err == nil || !strings.Contains(err.Error(), tt.err)) { + t.Errorf("err = %v, want %q", err, tt.err) + } + if n := reqs.get(seg); n != tt.requests { + t.Errorf("%d requests, want %d", n, tt.requests) + } + }) + } +} + +func TestSegmentsStall(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + stalled := segmentPath(t, src, "stream_0.m3u8", 1) + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == stalled { + reqs.add(stalled) + w.WriteHeader(http.StatusOK) + w.Write([]byte("x")) + w.(http.Flusher).Flush() + <-r.Context().Done() + return + } + http.FileServer(http.Dir(src)).ServeHTTP(w, r) + })) + defer srv.Close() + start := time.Now() + err := (&FFmpeg{segmentTimeout: 200 * time.Millisecond, retryDelay: noRetryDelay}).Download(context.Background(), + provider.Stream{URL: srv.URL + "/master.m3u8"}, filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) + if err == nil || !strings.Contains(err.Error(), "video segment 2 of 3: stalled: no data for 200ms") { + t.Errorf("err = %v", err) + } + if n := reqs.get(stalled); n != segmentTries { + t.Errorf("%d requests, want %d", n, segmentTries) + } + if elapsed := time.Since(start); elapsed > 10*time.Second { + t.Errorf("took %v", elapsed) + } +} + +func TestSegmentsInParallel(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + var running, most atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, ".m4s") { + n := running.Add(1) + defer running.Add(-1) + for m := most.Load(); n > m && !most.CompareAndSwap(m, n); m = most.Load() { + } + time.Sleep(100 * time.Millisecond) + } + http.FileServer(http.Dir(src)).ServeHTTP(w, r) + })) + defer srv.Close() + err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, + filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) + if err != nil { + t.Fatal(err) + } + if n := most.Load(); n < 2 || n > segmentWorkers { + t.Errorf("%d segments at once, want 2 to %d", n, segmentWorkers) + } +} + +// makeTS writes index.m3u8 with muxed TS segments, and master.m3u8 for it. +func makeTS(t *testing.T, dir string) { + t.Helper() + encoders, _ := exec.Command("ffmpeg", "-hide_banner", "-encoders").Output() + if !strings.Contains(string(encoders), "libx264") { + t.Skip("ffmpeg lacks libx264") + } + cmd := exec.Command("ffmpeg", "-hide_banner", "-loglevel", "error", + "-f", "lavfi", "-i", "testsrc=size=160x120:rate=10:duration=6", + "-f", "lavfi", "-i", "sine=duration=6", + "-c:v", "libx264", "-pix_fmt", "yuv420p", "-g", "10", "-c:a", "aac", + "-f", "hls", "-hls_time", "1", "-hls_playlist_type", "vod", + filepath.Join(dir, "index.m3u8")) + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("generating HLS: %v\n%s", err, out) + } + master := "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=300000,RESOLUTION=160x120,CODECS=\"avc1.64000c,mp4a.40.2\"\nindex.m3u8\n" + if err := os.WriteFile(filepath.Join(dir, "master.m3u8"), []byte(master), 0o644); err != nil { + t.Fatal(err) + } +} + +func TestSegmentsTS(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeTS(t, src) + srv := httptest.NewServer(http.FileServer(http.Dir(src))) + defer srv.Close() + work := t.TempDir() + out := filepath.Join(work, "out.mp4") + var bytes int64 + err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, + out, func(_ time.Duration, b int64) { bytes = b }) + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(work, segmentsDir, "v", "000005.ts")); err != nil { + t.Error(err) + } + fi, err := os.Stat(out) + if err != nil || bytes != fi.Size() { + t.Errorf("last progress %d bytes, file %v, %v", bytes, fi, err) + } + probe, err := exec.Command("ffprobe", "-v", "error", "-show_entries", "stream=codec_type", + "-of", "csv=p=0", out).Output() + if got := strings.Fields(string(probe)); err != nil || len(got) != 2 { + t.Errorf("streams = %v, %v; want video and audio", got, err) + } +} + +// A damaged segment fails the remux; fetching it again may help. +func TestSegmentsCorruptTS(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeTS(t, src) + seg := filepath.Join(src, strings.TrimPrefix(segmentPath(t, src, "index.m3u8", 2), "/")) + b, err := os.ReadFile(seg) + if err != nil { + t.Fatal(err) + } + const ts = 188 // TS packet size + if len(b) < 40*ts { + t.Fatalf("segment too small: %d bytes", len(b)) + } + if err := os.WriteFile(seg, append(b[:20*ts:20*ts], b[40*ts:]...), 0o644); err != nil { + t.Fatal(err) + } + srv := httptest.NewServer(http.FileServer(http.Dir(src))) + defer srv.Close() + work := t.TempDir() + err = (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, + filepath.Join(work, "out.mp4"), func(time.Duration, int64) {}) + if err == nil || !strings.Contains(err.Error(), "incomplete download") { + t.Fatalf("err = %v", err) + } + if _, err := os.Stat(filepath.Join(work, segmentsDir)); !os.IsNotExist(err) { + t.Errorf("damaged segments kept: %v", err) + } +} + +func TestSegmentExt(t *testing.T) { + for uri, want := range map[string]string{ + "https://cdn.example/a/seg-1.ts?token=x.mp4": ".ts", + "https://cdn.example/seg.AAC": ".aac", + "https://cdn.example/seg.ec3": ".eac3", + "https://cdn.example/seg.m4s": ".mp4", + "https://cdn.example/seg.php?id=1": ".ts", + "https://cdn.example/seg": ".ts", + } { + if got := segmentExt(hls.Segment{URI: uri}); got != want { + t.Errorf("%s: %q, want %q", uri, got, want) + } + } + if got := segmentExt(hls.Segment{URI: "https://cdn.example/seg.ts", Init: "https://cdn.example/init"}); got != ".mp4" { + t.Errorf("segment with an initialization section: %q, want .mp4", got) + } +} + +// Files are saved by position, so renamed ones are another stream's. +func TestSegmentsRestartForRenamedFiles(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + media, err := os.ReadFile(filepath.Join(src, "stream_0.m3u8")) + if err != nil { + t.Fatal(err) + } + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reqs.add(r.URL.Path) + name := filepath.Base(r.URL.Path) + switch { + case r.URL.Path == "/v2/stream_0.m3u8": + w.Write([]byte(strings.ReplaceAll(string(media), "stream_0", "fresh_0"))) + return + case strings.HasPrefix(name, "fresh_0"): + name = strings.Replace(name, "fresh_0", "stream_0", 1) + } + http.ServeFile(w, r, filepath.Join(src, name)) + })) + defer srv.Close() + out := filepath.Join(t.TempDir(), "out.mp4") + noop := func(time.Duration, int64) {} + for _, v := range []string{"/v1/", "/v2/"} { + if err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + v + "master.m3u8"}, out, noop); err != nil { + t.Fatal(err) + } + } + if reqs.get("/v2/fresh_00.m4s") != 1 || reqs.get("/v2/init_2.mp4") != 1 { + t.Error("kept the files of a stream whose segments are named differently") + } +} + +// A damaged initialization section fails the remux; it mustn't fail every retry. +func TestSegmentsRemoveFilesAfterFailedRemux(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + bad := "/init_0.mp4" + var broken atomic.Bool + broken.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == bad && broken.Load() { + w.Header().Set("Content-Type", "video/mp4") + w.Write([]byte("Service temporarily unavailable")) + return + } + http.FileServer(http.Dir(src)).ServeHTTP(w, r) + })) + defer srv.Close() + work := t.TempDir() + out := filepath.Join(work, "out.mp4") + s := provider.Stream{URL: srv.URL + "/master.m3u8"} + noop := func(time.Duration, int64) {} + if err := (&FFmpeg{}).Download(context.Background(), s, out, noop); err == nil { + t.Fatal("remuxed an error page") + } + if _, err := os.Stat(filepath.Join(work, segmentsDir)); !os.IsNotExist(err) { + t.Fatalf("damaged segments kept: %v", err) + } + broken.Store(false) + if err := (&FFmpeg{}).Download(context.Background(), s, out, noop); err != nil { + t.Fatalf("after the site recovered: %v", err) + } +} + +func TestMissingFFmpegFailsFirst(t *testing.T) { + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reqs.add("any") + http.NotFound(w, r) + })) + defer srv.Close() + err := (&FFmpeg{Path: filepath.Join(t.TempDir(), "ffmpeg")}).Download(context.Background(), + provider.Stream{URL: srv.URL + "/master.m3u8"}, filepath.Join(t.TempDir(), "out.mp4"), func(time.Duration, int64) {}) + if err == nil || !strings.HasPrefix(err.Error(), "ffmpeg: ") || reqs.get("any") != 0 { + t.Errorf("err = %v after %d requests", err, reqs.get("any")) + } +} + +// The output's failures leave the downloaded segments alone. +func TestSegmentsKeptAfterOutputFailure(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + srv := httptest.NewServer(http.FileServer(http.Dir(src))) + defer srv.Close() + work := t.TempDir() + out := filepath.Join(work, "out.mp4") + if err := os.Mkdir(out, 0o777); err != nil { // unwritable as a file + t.Fatal(err) + } + err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + "/master.m3u8"}, out, func(time.Duration, int64) {}) + if err == nil { + t.Fatal("wrote over a folder") + } + if _, err := os.Stat(filepath.Join(work, segmentsDir, "v", "000002.mp4")); err != nil { + t.Errorf("segments removed after %v", err) + } +} + +// Renditions may differ only in bitrate and share every file name. +func TestSegmentsRestartForAnotherBitrate(t *testing.T) { + needFFmpeg(t) + src := t.TempDir() + makeHLS(t, src) + var reqs requests + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reqs.add(r.URL.Path) + if bw, ok := strings.CutPrefix(r.URL.Path, "/master-"); ok { + dir := map[string]string{"200000.m3u8": "hi", "100000.m3u8": "lo"}[bw] + fmt.Fprintf(w, "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=%s,RESOLUTION=160x120,CODECS=\"avc1.64000c,mp4a.40.2\"\n%s/stream_0.m3u8\n", + strings.TrimSuffix(bw, ".m3u8"), dir) + return + } + http.ServeFile(w, r, filepath.Join(src, filepath.Base(r.URL.Path))) + })) + defer srv.Close() + out := filepath.Join(t.TempDir(), "out.mp4") + noop := func(time.Duration, int64) {} + for _, master := range []string{"/master-200000.m3u8", "/master-100000.m3u8"} { + if err := (&FFmpeg{}).Download(context.Background(), provider.Stream{URL: srv.URL + master}, out, noop); err != nil { + t.Fatal(err) + } + } + if reqs.get("/lo/stream_00.m4s") != 1 { + t.Error("kept the files of a rendition at another bitrate") + } +} diff --git a/internal/hls/hls.go b/internal/hls/hls.go index 289093e..4b0d1fc 100644 --- a/internal/hls/hls.go +++ b/internal/hls/hls.go @@ -1,4 +1,5 @@ -// Package hls selects video and audio from HLS master playlists. +// Package hls reads HLS playlists: it selects video and audio from master +// playlists and lists the segments of media playlists. package hls import ( @@ -44,7 +45,7 @@ func Load(ctx context.Context, client *http.Client, s provider.Stream) (m Master } playlist := s.Playlist if playlist == "" { - if playlist, base, err = fetch(ctx, client, s); err != nil { + if playlist, base, err = fetch(ctx, client, s.URL, s.Header); err != nil { return Master{}, false, fmt.Errorf("fetching playlist: %w", err) } } @@ -60,13 +61,16 @@ func Load(ctx context.Context, client *http.Client, s provider.Stream) (m Master return Master{Video: video, Audio: audio.uri, AudioLanguage: audio.language}, true, nil } +// Hours of short segments with tokenized URIs make media playlists this long. +const maxPlaylist = 16 << 20 + // Return the playlist and its URL after redirects. -func fetch(ctx context.Context, client *http.Client, s provider.Stream) (string, *url.URL, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.URL, nil) +func fetch(ctx context.Context, client *http.Client, uri string, h http.Header) (string, *url.URL, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, uri, nil) if err != nil { return "", nil, err } - for k, vs := range s.Header { + for k, vs := range h { for _, v := range vs { req.Header.Add(k, v) } @@ -79,10 +83,13 @@ func fetch(ctx context.Context, client *http.Client, s provider.Stream) (string, if resp.StatusCode < 200 || resp.StatusCode > 299 { return "", nil, fmt.Errorf("HTTP %d", resp.StatusCode) } - body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + body, err := io.ReadAll(io.LimitReader(resp.Body, maxPlaylist+1)) if err != nil { return "", nil, err } + if len(body) > maxPlaylist { + return "", nil, fmt.Errorf("over %d bytes", maxPlaylist) + } return string(body), resp.Request.URL, nil } diff --git a/internal/hls/media.go b/internal/hls/media.go new file mode 100644 index 0000000..659e19c --- /dev/null +++ b/internal/hls/media.go @@ -0,0 +1,93 @@ +package hls + +import ( + "bufio" + "context" + "errors" + "fmt" + "math" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// ErrUnsupported marks media playlists whose segments can't simply be fetched +// one by one, e.g. live or encrypted ones. +var ErrUnsupported = errors.New("unsupported media playlist") + +// Segment is one media segment of a VOD playlist. +type Segment struct { + URI string // absolute + Duration time.Duration + // Init is the absolute URI of the #EXT-X-MAP initialization section, "" if none. + Init string + // Timestamps may jump from the previous segment. + Discontinuity bool +} + +// LoadMedia fetches a VOD media playlist with header h and lists its segments. +func LoadMedia(ctx context.Context, client *http.Client, uri string, h http.Header) ([]Segment, error) { + playlist, base, err := fetch(ctx, client, uri, h) + if err != nil { + return nil, fmt.Errorf("fetching media playlist: %w", err) + } + return parseMedia(playlist, base) +} + +func parseMedia(playlist string, base *url.URL) ([]Segment, error) { + unsupported := func(why string) ([]Segment, error) { + return nil, fmt.Errorf("%w: %s", ErrUnsupported, why) + } + var segs []Segment + var next Segment + var init string + var timed, ended bool + sc := bufio.NewScanner(strings.NewReader(playlist)) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + tag, value, _ := strings.Cut(line, ":") + switch { + case line == "": + case tag == "#EXTINF": + d, _, _ := strings.Cut(value, ",") + secs, err := strconv.ParseFloat(strings.TrimSpace(d), 64) + if err != nil || !(secs >= 0) || math.IsInf(secs, 1) { + return unsupported(fmt.Sprintf("duration %q", d)) + } + next.Duration = time.Duration(math.Round(secs * float64(time.Second))) + timed = true + case tag == "#EXT-X-MAP": + a := parseAttrs(value) + if a["URI"] == "" || a["BYTERANGE"] != "" { + return unsupported("initialization section " + value) + } + init = resolve(base, a["URI"]) + case tag == "#EXT-X-KEY": + if a := parseAttrs(value); a["METHOD"] != "NONE" { + return unsupported("encrypted with " + a["METHOD"]) + } + case tag == "#EXT-X-DISCONTINUITY": + next.Discontinuity = true + case tag == "#EXT-X-ENDLIST": + ended = true + case tag == "#EXT-X-BYTERANGE", tag == "#EXT-X-GAP", tag == "#EXT-X-DEFINE", tag == "#EXT-X-STREAM-INF": + return unsupported(tag) + case strings.HasPrefix(line, "#"): + case !timed: + return unsupported("segment without a duration") + default: + next.URI, next.Init = resolve(base, line), init + segs = append(segs, next) + next, timed = Segment{}, false + } + } + switch { + case !ended: + return unsupported("live playlist") + case len(segs) == 0: + return unsupported("no segments") + } + return segs, nil +} diff --git a/internal/hls/media_test.go b/internal/hls/media_test.go new file mode 100644 index 0000000..5c99a28 --- /dev/null +++ b/internal/hls/media_test.go @@ -0,0 +1,133 @@ +package hls + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "strings" + "testing" + "time" +) + +// TVP's fMP4 media playlist (2026-10-01), shortened. +const tvpMedia = `#EXTM3U +#EXT-X-TARGETDURATION:4 +#EXT-X-ALLOW-CACHE:YES +#EXT-X-PLAYLIST-TYPE:VOD +#EXT-X-VERSION:6 +#EXT-X-MEDIA-SEQUENCE:1 +#EXT-X-MAP:URI="nv-dash-init-vod4-f5-v1-x3.mp4" +#EXTINF:4.000, +nv-dash-frag-vod4-1-f5-v1-x3.m4s +#EXTINF:3.840, +nv-dash-frag-vod4-2-f5-v1-x3.m4s +#EXT-X-ENDLIST +` + +// BBC's TS media playlist (2026-10-01), shortened. +const bbcMedia = `#EXTM3U +#EXT-X-VERSION:2 +## Created with Unified Streaming Platform(version=1.7.32) +#EXT-X-MEDIA-SEQUENCE:1 +#EXT-X-TARGETDURATION:8 +#USP-X-TIMESTAMP-MAP:MPEGTS=900000,LOCAL=1970-01-01T00:00:00Z +#EXTINF:8, no desc +vf.ism.hlsv2-audio_eng_1=128000-video=12000000-1.ts +#EXTINF:7.08, no desc +vf.ism.hlsv2-audio_eng_1=128000-video=12000000-2.ts +#EXT-X-ENDLIST +` + +func TestParseMedia(t *testing.T) { + base, _ := url.Parse("https://cdn.example/token/abc/video.ism/index.m3u8?t=1") + dir := "https://cdn.example/token/abc/video.ism/" + tests := []struct { + name, playlist string + want []Segment + }{ + {"tvp", tvpMedia, []Segment{ + {URI: dir + "nv-dash-frag-vod4-1-f5-v1-x3.m4s", Duration: 4 * time.Second, Init: dir + "nv-dash-init-vod4-f5-v1-x3.mp4"}, + {URI: dir + "nv-dash-frag-vod4-2-f5-v1-x3.m4s", Duration: 3840 * time.Millisecond, Init: dir + "nv-dash-init-vod4-f5-v1-x3.mp4"}, + }}, + {"bbc", bbcMedia, []Segment{ + {URI: dir + "vf.ism.hlsv2-audio_eng_1=128000-video=12000000-1.ts", Duration: 8 * time.Second}, + {URI: dir + "vf.ism.hlsv2-audio_eng_1=128000-video=12000000-2.ts", Duration: 7080 * time.Millisecond}, + }}, + {"discontinuity, new map, absolute URIs, no encryption", `#EXTM3U +#EXT-X-KEY:METHOD=NONE +#EXT-X-MAP:URI="init1.mp4" +#EXTINF:2, +https://other.example/1.m4s?sig=x +#EXT-X-DISCONTINUITY +#EXT-X-MAP:URI="/init2.mp4" +#EXTINF:2.5, +2.m4s +#EXT-X-ENDLIST +`, []Segment{ + {URI: "https://other.example/1.m4s?sig=x", Duration: 2 * time.Second, Init: dir + "init1.mp4"}, + {URI: dir + "2.m4s", Duration: 2500 * time.Millisecond, Init: "https://cdn.example/init2.mp4", Discontinuity: true}, + }}, + } + for _, tt := range tests { + got, err := parseMedia(tt.playlist, base) + if err != nil || !reflect.DeepEqual(got, tt.want) { + t.Errorf("%s: got %+v, %v; want %+v", tt.name, got, err, tt.want) + } + } + + for name, playlist := range map[string]string{ + "live": "#EXTM3U\n#EXTINF:4,\n1.ts\n", + "empty": "#EXTM3U\n#EXT-X-ENDLIST\n", + "AES-128": "#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"k\"\n#EXTINF:4,\n1.ts\n#EXT-X-ENDLIST\n", + "SAMPLE-AES": "#EXTM3U\n#EXT-X-KEY:METHOD=SAMPLE-AES,URI=\"skd://k\"\n#EXTINF:4,\n1.ts\n#EXT-X-ENDLIST\n", + "byte range": "#EXTM3U\n#EXTINF:4,\n#EXT-X-BYTERANGE:100@0\nall.ts\n#EXT-X-ENDLIST\n", + "ranged map": "#EXTM3U\n#EXT-X-MAP:URI=\"all.mp4\",BYTERANGE=\"100@0\"\n#EXTINF:4,\n1.m4s\n#EXT-X-ENDLIST\n", + "gap": "#EXTM3U\n#EXTINF:4,\n#EXT-X-GAP\n1.ts\n#EXT-X-ENDLIST\n", + "variables": "#EXTM3U\n#EXT-X-DEFINE:NAME=\"p\",VALUE=\"x\"\n#EXTINF:4,\n{$p}.ts\n#EXT-X-ENDLIST\n", + "master": "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\nv.m3u8\n", + "no duration": "#EXTM3U\n1.ts\n#EXT-X-ENDLIST\n", + "bad duration": "#EXTM3U\n#EXTINF:NaN,\n1.ts\n#EXT-X-ENDLIST\n", + "not a playlist": "Sorry\n", + "negative length": "#EXTM3U\n#EXTINF:-1,\n1.ts\n#EXT-X-ENDLIST\n", + } { + if segs, err := parseMedia(playlist, base); !errors.Is(err, ErrUnsupported) { + t.Errorf("%s: got %+v, %v; want ErrUnsupported", name, segs, err) + } + } +} + +func TestLoadMedia(t *testing.T) { + var gotUA string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotUA = r.UserAgent() + switch r.URL.Path { + case "/old/index.m3u8": + http.Redirect(w, r, "/new/index.m3u8", http.StatusFound) + case "/new/index.m3u8": + io.WriteString(w, bbcMedia) + case "/huge.m3u8": + io.WriteString(w, strings.Repeat("#", 16<<20+1)) + default: + http.Error(w, "forbidden", http.StatusForbidden) + } + })) + defer srv.Close() + ctx := context.Background() + + segs, err := LoadMedia(ctx, srv.Client(), srv.URL+"/old/index.m3u8", http.Header{"User-Agent": {"magnetowid-test"}}) + if err != nil || len(segs) != 2 || segs[0].URI != srv.URL+"/new/vf.ism.hlsv2-audio_eng_1=128000-video=12000000-1.ts" || + gotUA != "magnetowid-test" { + t.Errorf("got %+v, %v (UA %q)", segs, err, gotUA) + } + if _, err := LoadMedia(ctx, srv.Client(), srv.URL+"/refused.m3u8", nil); err == nil || + err.Error() != "fetching media playlist: HTTP 403" { + t.Errorf("refused: err = %v", err) + } + if _, err := LoadMedia(ctx, srv.Client(), srv.URL+"/huge.m3u8", nil); err == nil || errors.Is(err, ErrUnsupported) { + t.Errorf("oversized: err = %v", err) + } +} diff --git a/internal/web/templates/queue.html b/internal/web/templates/queue.html index b3acc48..d8954aa 100644 --- a/internal/web/templates/queue.html +++ b/internal/web/templates/queue.html @@ -5,7 +5,7 @@ {{- else -}}
- +
{{- end}} {{- end}} @@ -82,14 +82,14 @@

The queue is empty

{{- else}}
- +
{{- end}} {{template "remove-button" .}}