From 1b29b1de03ff4d7dc111c0844ff42077984dd395 Mon Sep 17 00:00:00 2001 From: Piotr Komborski <163394+combor@users.noreply.github.com> Date: Thu, 1 Oct 2026 10:30:08 +0100 Subject: [PATCH 1/2] Read the segments of HLS media playlists LoadMedia lists a VOD media playlist's segments with their durations, initialization sections and discontinuities, resolving URIs against the playlist's URL after redirects. Playlists whose segments can't simply be fetched one by one return ErrUnsupported: live and encrypted ones, and those with byte ranges, gaps or variables. Neither BBC nor TVP uses these. Playlists may now be up to 16 MiB, as hours of short segments with tokenized URIs exceed the old 1 MiB, and longer ones fail instead of being cut short. --- internal/hls/hls.go | 19 ++++-- internal/hls/media.go | 93 ++++++++++++++++++++++++++ internal/hls/media_test.go | 133 +++++++++++++++++++++++++++++++++++++ 3 files changed, 239 insertions(+), 6 deletions(-) create mode 100644 internal/hls/media.go create mode 100644 internal/hls/media_test.go 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) + } +} From 12f8e290b7af075ab145deec2ac3d3ad790d0be7 Mon Sep 17 00:00:00 2001 From: Piotr Komborski <163394+combor@users.noreply.github.com> Date: Thu, 1 Oct 2026 10:30:09 +0100 Subject: [PATCH 2/2] Continue paused and interrupted downloads where they stopped Pausing a download, restarting magnetowid or retrying a failed attempt started the download over: ffmpeg wrote the MP4 in one pass, and the queue deleted the partial file. magnetowid now fetches HLS segments itself, four at a time, into the job's folder in .incomplete, and ffmpeg only remuxes the saved files. A resumed download looks the stream up again, since its URLs expire, and skips the segments already saved. A segment appears under its final name only once it is complete. Segments are saved by position, because their URLs carry tokens that change with every lookup. The local playlists ffmpeg reads name the rendition, its bandwidth and a hash of the segments' file names, so if the site serves a different stream, the saved segments are discarded and the download starts over. A segment that fails with a server error, a rate limit, a stall or an error page is fetched up to five times. Other client errors, such as an expired token, fail the attempt, and the queue's retry resolves fresh URLs and continues. If ffmpeg finds a saved file damaged or unreadable, the segments are discarded so that retries fetch them again; other remux failures, such as a full disk, keep them. The queue keeps a job's folder through pauses, restarts and retries, and removes it when the job finishes, fails for good or is removed. Paused jobs keep their progress, shown as "Paused at 40%", and the time left counts only the current attempt's progress. Live, encrypted and non-HLS streams still go straight to ffmpeg and start over. Finishing a download needs free space for about twice the video's size. --- cmd/magnetowid/main.go | 2 +- docs/usage.md | 22 +- internal/downloader/ffmpeg.go | 172 ++++++-- internal/downloader/ffmpeg_test.go | 22 +- internal/downloader/ffmpeg_unix_test.go | 86 ++++ internal/downloader/queue.go | 87 ++-- internal/downloader/queue_test.go | 212 +++++++++- internal/downloader/segments.go | 447 ++++++++++++++++++++ internal/downloader/segments_test.go | 515 ++++++++++++++++++++++++ internal/web/templates/queue.html | 6 +- internal/web/view.go | 3 + internal/web/web_test.go | 6 +- 12 files changed, 1490 insertions(+), 90 deletions(-) create mode 100644 internal/downloader/segments.go create mode 100644 internal/downloader/segments_test.go 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/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" .}}