httpdl: fix progress-reporter race on name/out; harden mirror & resume validation

Findings from a Rob-Pike-lens review (bugs/races/network), each verified
against the code before fixing:

- httpdl: name/out are now atomic.Pointer[string] -- the engine publishes a
  download to the reporter before Run resolves the name, so Stat raced the
  write (bt already did this)
- httpdl: a malformed --proxy fails loudly instead of silently bypassing it
- httpdl: a 206 must carry a matching Content-Range; a 200 in segmented mode is
  fatal so it fails over instead of burning the retry budget
- httpdl: single-stream mirror failover validates the range before appending;
  ErrTooSlow only when the error is a real ctx cancellation
- httpdl: idle guard tracks progress by timestamp (no Reset/Stop race, no
  sticky fired flag)
- httpdl/control: reject a resume file whose segments don't tile [0,total)
- bt: clamp the listen-port range; verify on-disk data before choosing pieces
  under --check-integrity
- cli: reject size overflow; show --seed-time=MIN; clamp --select-file range
- progress/engine/main: clamp ETA against int64 overflow; show queued
  downloads as waiting; join the reporter on exit instead of a 20ms sleep

httpdl/control.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"os"
 	"path/filepath"
+	"sort"
 	"sync/atomic"
 )
 
@@ -75,7 +76,35 @@ func loadControl(out, url string, total int64, etag, lastmod string) *control {
 	if (lastmod != "" || c.LastModified != "") && c.LastModified != lastmod {
 		return nil
 	}
+	// Reject a control file whose segments do not exactly tile [0,total): a
+	// truncated/corrupted/hand-edited sidecar that still parses as JSON could
+	// otherwise mark a segment done() without its bytes on disk (inflated Written)
+	// or leave an un-downloaded hole, both of which would be reported as a complete
+	// file. We restart cleanly instead of trusting it.
+	if !validSegs(c.Segs, c.Total) {
+		return nil
+	}
 	return &c
 }
 
+// validSegs reports whether segs cover [0,total) with no gap or overlap and a
+// sane written count for each. The recorded order is not sorted (a steal appends
+// a tail), so we check coverage on a sorted copy.
+func validSegs(segs []segState, total int64) bool {
+	if total <= 0 || len(segs) == 0 {
+		return false
+	}
+	sorted := append([]segState(nil), segs...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Start < sorted[j].Start })
+	var next int64
+	for _, s := range sorted {
+		length := s.End - s.Start + 1
+		if s.Start != next || s.End < s.Start || s.Written < 0 || s.Written > length {
+			return false
+		}
+		next = s.End + 1
+	}
+	return next == total
+}
+
 func removeControl(out string) { os.Remove(controlPath(out)) }

httpdl/httpdl.go

@@ -127,12 +127,13 @@ type Download struct {
 	jar      *cookiejar.Jar // non-nil when cookies are in use, for save-cookies
 
 	// name and out are resolved in Run after the probe response is known (the
-	// final post-redirect URL and any Content-Disposition can change them), then
-	// never mutated again, so Stat can read name from another goroutine without a
-	// lock once Run has set it. They are seeded in New so a Stat before Run still
-	// has a sensible best-effort name.
-	name    string
-	out     string
+	// final post-redirect URL and any Content-Disposition can change them). The
+	// engine publishes a Download to the progress reporter before Run starts, so
+	// Stat reads these from the reporter goroutine while resolveName writes them;
+	// they are atomic.Pointer[string] (exactly like bt.Download.name) to keep that
+	// read/write race-free. Seeded in New so a Stat before Run has a best-effort name.
+	name    atomic.Pointer[string]
+	out     atomic.Pointer[string]
 	initErr error
 
 	// mu guards seenURLs, which records every URL we sent a request to so
@@ -232,7 +233,21 @@ func New(uris []string, cfg Config) *Download {
 		TLSClientConfig:       tlsCfg,
 	}
 	if cfg.Proxy != "" {
-		if pu, err := url.Parse(cfg.Proxy); err == nil {
+		pu, err := url.Parse(cfg.Proxy)
+		switch {
+		case err != nil:
+			// A malformed proxy must fail loudly, not silently fall back to the
+			// environment proxy (or a direct connection) and leak the real IP.
+			if initErr == nil {
+				initErr = fmt.Errorf("proxy %q: %w", cfg.Proxy, err)
+			}
+		case pu.Host == "":
+			// "host:port" with no scheme parses to an opaque URL with no Host, which
+			// http.ProxyURL would route nowhere; require an explicit scheme.
+			if initErr == nil {
+				initErr = fmt.Errorf("proxy %q: missing scheme (use scheme://host:port)", cfg.Proxy)
+			}
+		default:
 			tr.Proxy = http.ProxyURL(pu)
 		}
 	}
@@ -294,29 +309,47 @@ func New(uris []string, cfg Config) *Download {
 		client.Jar = jar
 	}
 
-	return &Download{
+	d := &Download{
 		uris:     uris,
 		maxConns: maxConns,
 		cfg:      cfg,
 		client:   client,
 		limit:    limit,
 		jar:      jar,
-		name:     name,
-		out:      out,
 		total:    -1,
 		initErr:  initErr,
 	}
+	d.setName(name)
+	d.setOut(out)
+	return d
 }
 
-func (d *Download) Name() string { return d.name }
+func (d *Download) Name() string { return d.loadName() }
 
 // Path returns the resolved output file path (used by --follow-torrent).
-func (d *Download) Path() string { return d.out }
+func (d *Download) Path() string { return d.loadOut() }
+
+// name and out are read by Stat from the reporter goroutine while Run resolves
+// them, so they are stored atomically (mirroring bt.Download.name).
+func (d *Download) setName(s string) { d.name.Store(&s) }
+func (d *Download) setOut(s string)  { d.out.Store(&s) }
+func (d *Download) loadName() string {
+	if p := d.name.Load(); p != nil {
+		return *p
+	}
+	return ""
+}
+func (d *Download) loadOut() string {
+	if p := d.out.Load(); p != nil {
+		return *p
+	}
+	return ""
+}
 
 func (d *Download) Stat() download.Stat {
 	return download.Stat{
-		Name:      d.name,
-		ID:        d.out, // resolved output path; stable per process (CONTRACT)
+		Name:      d.loadName(),
+		ID:        d.loadOut(), // resolved output path; stable per process (CONTRACT)
 		IsBT:      false,
 		Status:    download.Status(atomic.LoadInt32(&d.status)),
 		Total:     atomic.LoadInt64(&d.total),
@@ -400,15 +433,19 @@ func (d *Download) Run(ctx context.Context) (err error) {
 	// the limit past a startup grace. The watcher cancels xferCtx and sets a
 	// flag so we can return the distinct "too slow" error rather than a bare
 	// cancellation.
+	out := d.loadOut()
 	xferCtx, stopGuard, tooSlow := d.speedGuard(ctx)
 	if !pr.ranged || pr.total <= 0 {
-		err = d.single(xferCtx, d.out)
+		err = d.single(xferCtx, out)
 	} else {
-		err = d.segmented(xferCtx, d.out, pr.total, pr.etag, pr.lastmod)
+		err = d.segmented(xferCtx, out, pr.total, pr.etag, pr.lastmod)
 	}
 	stopGuard()
 	if err != nil {
-		if tooSlow() {
+		// Only translate to ErrTooSlow when the transfer actually ended in the
+		// guard's cancellation; a write error or mirror exhaustion that merely
+		// raced the guard firing keeps its own (real) cause and exit code.
+		if tooSlow() && errors.Is(err, context.Canceled) {
 			return fmt.Errorf("%s: %w (<= %d bytes/sec)", d.primary(), ErrTooSlow, d.cfg.LowestSpeedLimit)
 		}
 		return err
@@ -416,14 +453,14 @@ func (d *Download) Run(ctx context.Context) (err error) {
 	// --checksum: verify the finished file before declaring success, so a
 	// corrupted or tampered download fails (--checksum, exit code 32).
 	if newSum != nil {
-		if err = verifyFile(d.out, newSum, sumWant); err != nil {
+		if err = verifyFile(out, newSum, sumWant); err != nil {
 			return err
 		}
 	}
 	// -R / --remote-time: stamp the finished file with the server's mtime.
 	if d.cfg.RemoteTime && pr.lastmod != "" {
 		if t, err := http.ParseTime(pr.lastmod); err == nil {
-			os.Chtimes(d.out, t, t)
+			os.Chtimes(out, t, t)
 		}
 	}
 	d.setStatus(download.Complete)
@@ -435,6 +472,7 @@ func (d *Download) Run(ctx context.Context) (err error) {
 // the original URL. Overwrite / auto-rename is then resolved against the final
 // path, honouring -c so a resumable file is never renamed (items [2],[8],[13]).
 func (d *Download) resolveName(pr probeResult) error {
+	out := d.loadOut()
 	if d.cfg.Out == "" {
 		name := ""
 		if cd := pr.cdisp; cd != "" {
@@ -446,8 +484,9 @@ func (d *Download) resolveName(pr probeResult) error {
 		if name == "" {
 			name = nameFromURL(d.primary())
 		}
-		d.name = name
-		d.out = filepath.Join(d.cfg.Dir, name)
+		out = filepath.Join(d.cfg.Dir, name)
+		d.setName(name)
+		d.setOut(out)
 	}
 
 	// --dry-run writes nothing, so skip overwrite/auto-rename resolution; the name
@@ -458,20 +497,20 @@ func (d *Download) resolveName(pr probeResult) error {
 
 	// -c resumes either our own .got sidecar or a foreign/browser partial file,
 	// so an existing file is never auto-renamed under -c (item [2]).
-	if d.cfg.Continue && fileExists(d.out) {
+	if d.cfg.Continue && fileExists(out) {
 		return nil
 	}
-	if fileExists(d.out) && !d.cfg.AllowOverwrite {
+	if fileExists(out) && !d.cfg.AllowOverwrite {
 		if d.cfg.AutoRename {
-			u, err := uniqueName(d.out)
+			u, err := uniqueName(out)
 			if err != nil {
 				return err
 			}
-			d.out = u
-			d.name = filepath.Base(u)
+			d.setOut(u)
+			d.setName(filepath.Base(u))
 			return nil
 		}
-		return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", d.out, ErrOutputExists)
+		return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", out, ErrOutputExists)
 	}
 	return nil
 }
@@ -587,8 +626,8 @@ func (d *Download) probe(ctx context.Context, uri string) (probeResult, error) {
 	// is already current. Only done when there is no .got control file (an
 	// in-progress resume must not be short-circuited), using the local file's
 	// mtime for If-Modified-Since.
-	if d.cfg.ConditionalGet && !fileExists(controlPath(d.out)) {
-		if fi, err := os.Stat(d.out); err == nil {
+	if out := d.loadOut(); d.cfg.ConditionalGet && !fileExists(controlPath(out)) {
+		if fi, err := os.Stat(out); err == nil {
 			req.Header.Set("If-Modified-Since", fi.ModTime().UTC().Format(http.TimeFormat))
 		}
 	}
@@ -757,60 +796,94 @@ func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, u
 		atomic.AddInt32(&d.conns, -1)
 	}()
 	if resp.StatusCode != http.StatusPartialContent {
-		return statusError(fmt.Sprintf("segment %d: expected 206, got %s", s.index, resp.Status), resp.StatusCode)
+		err := statusError(fmt.Sprintf("segment %d: expected 206, got %s", s.index, resp.Status), resp.StatusCode)
+		// A 200 means this mirror ignores Range entirely; retrying it only burns the
+		// budget, so make it fatal and let fetchSeg fail over to the next mirror.
+		if resp.StatusCode == http.StatusOK {
+			err = fatal{err}
+		}
+		return err
 	}
 	// A mirror must serve the same file as the primary, or its bytes written at
-	// this offset would corrupt the output. Reject a 206 whose total length
-	// disagrees with the probe (validate the total length); fatal so withRetries
-	// stops and fetchSeg falls over to the next mirror instead of writing it.
-	if _, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok && t != total {
-		return fatal{fmt.Errorf("segment %d: %s reports length %d, want %d", s.index, uri, t, total)}
+	// this offset would corrupt the output. Require a Content-Range that confirms
+	// both the offset we asked for and the total length the probe saw; a missing,
+	// unparseable, or divergent range is fatal so withRetries stops and fetchSeg
+	// falls over to the next mirror instead of writing unverified bytes.
+	if start, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); !ok || t != total || start != s.offset() {
+		return fatal{fmt.Errorf("segment %d: %s returned an unverifiable range %q (want start %d, length %d)",
+			s.index, uri, resp.Header.Get("Content-Range"), s.offset(), total)}
 	}
 	body, stop := d.idleGuard(resp.Body, cancel)
 	defer stop()
 	return d.pump(ctx, f, p, s, body)
 }
 
-// errIdleTimeout marks a read that stalled past the idle window, so callers can
-// report a distinct "timed out" instead of the bare "context canceled" that the
-// cancelled request would otherwise surface (item [39]).
+// errIdleTimeout marks a transfer that stalled past the idle window, so callers
+// can report a distinct "timed out" instead of the bare "context canceled" that
+// the cancelled request would otherwise surface (item [39]).
 var errIdleTimeout = errors.New("idle timeout: no data received")
 
-// idleReader resets a watchdog timer on every read; if a read stalls longer
-// than the idle window the timer fires and cancels the request, so the blocked
-// Read returns an error instead of hanging forever on a wedged connection.
+// idleReader records the time of the last byte received; a background watcher
+// (idleGuard) cancels the request once a full idle window passes with no
+// progress. Tracking progress by timestamp — rather than arming a per-read timer
+// — means a slow-but-advancing transfer is never aborted, and there is no
+// Reset/Stop race that could poison a connection that was still delivering data.
 type idleReader struct {
-	r     io.Reader
-	timer *time.Timer
-	idle  time.Duration
-	fired atomic.Bool // set when the watchdog cancelled the request
+	r        io.Reader
+	idle     time.Duration
+	last     atomic.Int64 // UnixNano of the last byte received
+	timedOut atomic.Bool  // set when the watcher cancelled the request
 }
 
 func (ir *idleReader) Read(p []byte) (int, error) {
-	ir.timer.Reset(ir.idle)
 	n, err := ir.r.Read(p)
-	ir.timer.Stop() // only the Read itself counts as idle, not write/throttle gaps
-	if err != nil && ir.fired.Load() {
-		// The cancellation came from our watchdog, not the caller's ctx, so
+	if n > 0 {
+		ir.last.Store(time.Now().UnixNano())
+	}
+	if err != nil && ir.timedOut.Load() {
+		// The cancellation came from our watcher, not the caller's ctx, so
 		// translate the read error into the distinct idle-timeout sentinel.
 		return n, fmt.Errorf("%w (after %s)", errIdleTimeout, ir.idle)
 	}
 	return n, err
 }
 
-// idleGuard wraps body with an idle timeout of d.cfg.Timeout, cancelling the
-// request if no data arrives in that window. It returns the reader to use and a
-// stop func to call when the transfer is done.
+// watch cancels the request once idle passes with no byte received. It polls a
+// few times per window so a stall is caught within ~idle of the last byte,
+// without a shared timer that Read would have to Reset/Stop.
+func (ir *idleReader) watch(stop <-chan struct{}, cancel context.CancelFunc) {
+	tick := ir.idle / 4
+	if tick <= 0 {
+		tick = ir.idle
+	}
+	t := time.NewTicker(tick)
+	defer t.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case now := <-t.C:
+			if now.UnixNano()-ir.last.Load() >= int64(ir.idle) {
+				ir.timedOut.Store(true)
+				cancel()
+				return
+			}
+		}
+	}
+}
+
+// idleGuard wraps body with an idle timeout of d.cfg.Timeout: a watcher goroutine
+// cancels the request when no byte arrives for that long. It returns the reader to
+// use and a stop func to call (exactly once) when the transfer is done.
 func (d *Download) idleGuard(body io.Reader, cancel context.CancelFunc) (io.Reader, func()) {
 	if d.cfg.Timeout <= 0 {
 		return body, func() {}
 	}
 	ir := &idleReader{r: body, idle: d.cfg.Timeout}
-	ir.timer = time.AfterFunc(d.cfg.Timeout, func() {
-		ir.fired.Store(true)
-		cancel()
-	})
-	return ir, func() { ir.timer.Stop() }
+	ir.last.Store(time.Now().UnixNano())
+	stop := make(chan struct{})
+	go ir.watch(stop, cancel)
+	return ir, func() { close(stop) }
 }
 
 // speedStartupGrace is how long a download may stay slow before the
@@ -1009,6 +1082,19 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
 	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
 		return statusError(fmt.Sprintf("%s: %s", uri, resp.Status), resp.StatusCode)
 	}
+	// On failover, resumeAt is the prefix a previous mirror wrote. If this mirror
+	// honours the range, make sure it serves the same file before we append to that
+	// prefix — a divergent total or start means different content, and writing it
+	// past the existing bytes would corrupt the output. Mirror the segmented guard
+	// (fatal, so we fall over rather than append unverified bytes). A same-mirror
+	// retry compares against its own earlier total, so it never false-trips.
+	if resp.StatusCode == http.StatusPartialContent && resumeAt > 0 {
+		if prev := atomic.LoadInt64(&d.total); prev > 0 {
+			if start, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok && (t != prev || start != resumeAt) {
+				return fatal{fmt.Errorf("%s: returned a divergent range %q (want start %d, length %d)", uri, resp.Header.Get("Content-Range"), resumeAt, prev)}
+			}
+		}
+	}
 	// If we asked to resume but the server replied 200 (no range honoured), start
 	// over from the top rather than appending past the existing bytes.
 	if resp.StatusCode == http.StatusOK {