Continue paused and interrupted downloads where they stopped

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
 	}

docs/usage.md

@@ -159,18 +159,18 @@ 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
 
 Open `http://<host>: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/<job id>` 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
 

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,16 +207,22 @@ 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)
 		sc := bufio.NewScanner(stderr)
 		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

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