got/httpdl/httpdl.go

// Package httpdl downloads a single HTTP(S) resource over one or more
// connections. The model is deliberately flat: split the file into byte-range
// segments, give each segment its own worker goroutine, and have each worker
// stream its range straight to disk with WriteAt (safe for concurrent,
// non-overlapping writes — no shared seek, no mutex). Goroutines blocking on
// real I/O keep the model simple: no segment manager, no piece storage, no
// command reactor.
package httpdl

import (
	"context"
	"errors"
	"fmt"
	"hash"
	"io"
	"mime"
	"net"
	"net/http"
	"net/url"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/hanbok/got/download"
	"golang.org/x/time/rate"
)

const readBuf = 32 * 1024

// minSplitSize is the smallest piece a file is split into: a segment shorter
// than this is not worth its own connection, so makeSegments stops dividing once
// the pieces would fall below it.
const minSplitSize = 20 << 20

// userAgent is sent on every request; there is no flag to change it. A site that
// needs a different one can be given a User-Agent header via --header.
const userAgent = "got/1.0"

// ErrOutputExists reports that the resolved output file already exists and
// neither --allow-overwrite nor -c was given. main maps it to a distinct exit
// code via errors.Is, so it must stay on the error chain (CONTRACT).
var ErrOutputExists = errors.New("output file already exists")

// ErrNotFound reports a permanent "not found" HTTP status (e.g. 404). main maps
// it to a distinct exit code via errors.Is, so it must stay on the error chain
// (CONTRACT).
var ErrNotFound = errors.New("not found")

// Config holds everything one HTTP download needs. It is assembled by main from
// the resolved CLI options.
type Config struct {
	Dir            string
	Out            string
	Split          int   // --split: total connections across all mirrors
	MinSplit       int64 // smallest piece to split into; 0 = minSplitSize default (internal, no flag)
	Tries          int   // --max-tries: 0 = unlimited
	Timeout        time.Duration
	FileAlloc      string // none | prealloc | trunc | falloc
	Continue       bool
	AllowOverwrite bool
	AutoRename     bool
	Headers        []string      // --header: extra request headers
	OverallLimiter *rate.Limiter // shared across all downloads, may be nil
	// Checksum is a "<type>=<digest>" spec (e.g. "sha-256=ab…"); when set,
	// the finished file is hashed and compared, failing with ErrChecksum on a
	// mismatch (--checksum). Empty means no verification.
	Checksum string
	// DryRun probes the resource — proving it exists and its size — but downloads
	// nothing, then reports success (--dry-run).
	DryRun bool
}

// Download implements download.Download for one HTTP(S) file, fetched from one
// or more mirror URLs that serve identical bytes. uris[0] is the primary — it
// names the file and keys the resume sidecar; the rest are fallbacks/extra
// connections.
type Download struct {
	uris     []string // mirror list; uris[0] is the primary (naming + resume key)
	maxConns int      // segment-worker cap = min(split, len(uris))
	cfg      Config
	client   *http.Client
	limit    []*rate.Limiter

	// 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). 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

	// live counters, read by Stat from any goroutine
	total     int64 // -1 until known
	completed int64
	status    int32 // download.Status
	conns     int32 // active connections
}

// primary is the canonical URL — used for naming and the resume sidecar so they
// stay stable regardless of which mirror served the bytes.
func (d *Download) primary() string { return d.uris[0] }

// mirror returns the URL for the n-th pick. Choosing by a caller-supplied index
// (segment index, retry attempt) spreads connections across the mirrors and
// rotates on retry, with no shared state — so it stays lock-free.
func (d *Download) mirror(n int) string { return d.uris[n%len(d.uris)] }

// New builds an HTTP download for uris, which must all serve identical content
// (mirrors). uris[0] is the primary.
func New(uris []string, cfg Config) *Download {
	var initErr error
	if cfg.MinSplit <= 0 {
		cfg.MinSplit = minSplitSize
	}

	// One connect budget: --timeout bounds both establishing the connection and
	// idle stalls once data flows. Proxies still come from the standard
	// HTTP_PROXY/HTTPS_PROXY environment, and TLS uses the system roots.
	tr := &http.Transport{
		Proxy:                 http.ProxyFromEnvironment,
		MaxConnsPerHost:       cfg.Split,
		MaxIdleConnsPerHost:   cfg.Split,
		ResponseHeaderTimeout: cfg.Timeout,
		TLSHandshakeTimeout:   cfg.Timeout,
		DialContext:           (&net.Dialer{Timeout: cfg.Timeout}).DialContext,
	}

	var limit []*rate.Limiter
	if cfg.OverallLimiter != nil {
		limit = append(limit, cfg.OverallLimiter)
	}

	// Seed a best-effort name from -o or the primary URL. The real name is
	// finalised in Run after probe(), where Content-Disposition and the
	// post-redirect URL are known (item [8]); name and out stay immutable then on.
	primary := ""
	if len(uris) > 0 {
		primary = uris[0]
	}
	name := cfg.Out
	if name == "" {
		name = nameFromURL(primary)
	}
	out := filepath.Join(cfg.Dir, name)

	// An empty mirror list can download nothing; surface it as an initErr so Run
	// returns cleanly instead of indexing uris[0].
	if len(uris) == 0 && initErr == nil {
		initErr = errors.New("no URL to download")
	}
	// Worker count is --split: the file is cut into that many byte-range segments,
	// each its own connection, whether served by one host or spread across mirrors.
	maxConns := cfg.Split
	if maxConns < 1 {
		maxConns = 1
	}

	d := &Download{
		uris:     uris,
		maxConns: maxConns,
		cfg:      cfg,
		client:   &http.Client{Transport: tr},
		limit:    limit,
		total:    -1,
		initErr:  initErr,
	}
	d.setName(name)
	d.setOut(out)
	return d
}

func (d *Download) Name() string { return d.loadName() }

// Path returns the resolved output file path, so a fetched .torrent can be
// followed into a BitTorrent download of its content.
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.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),
		Completed: atomic.LoadInt64(&d.completed),
		Conns:     int(atomic.LoadInt32(&d.conns)),
	}
}

// setStatus stores the lifecycle state with a single atomic op so Stat stays
// lock-free.
func (d *Download) setStatus(s download.Status) {
	atomic.StoreInt32(&d.status, int32(s))
}

// probeResult carries everything probe() learned, so name finalisation can run
// after the request (the final URL and Content-Disposition are only known then).
type probeResult struct {
	total    int64
	ranged   bool
	etag     string
	lastmod  string
	cdisp    string // raw Content-Disposition header
	finalURL *url.URL
}

// Run probes the resource, finalises the output name, decides between a single
// stream and a segmented download, transfers the bytes, and finalises the
// resume file. Errored is set exactly once via the deferred check of err.
func (d *Download) Run(ctx context.Context) (err error) {
	// One place sets Errored, keyed on whatever err we return (item [37]).
	defer func() {
		if err != nil {
			d.setStatus(download.Errored)
		}
	}()

	if d.initErr != nil {
		return d.initErr
	}
	d.setStatus(download.Active)

	pr, err := d.probeRetry(ctx)
	if err != nil {
		return err
	}
	atomic.StoreInt64(&d.total, pr.total)

	// Finalise name/out now that the post-redirect URL and Content-Disposition
	// are known (item [8]), then resolve overwrite/auto-rename (items [2],[13]).
	if err = d.resolveName(pr); err != nil {
		return err
	}

	// --checksum: parse the spec up front so a bad type or digest fails before we
	// download rather than after. The finished file is verified below.
	var (
		newSum  func() hash.Hash
		sumWant string
	)
	if d.cfg.Checksum != "" {
		if newSum, sumWant, err = parseChecksum(d.cfg.Checksum); err != nil {
			return err
		}
	}

	// --dry-run: the probe already proved the resource exists and (above) its
	// size, which is all a dry run needs — stop here without downloading.
	if d.cfg.DryRun {
		d.setStatus(download.Complete)
		return nil
	}

	out := d.loadOut()
	if !pr.ranged || pr.total <= 0 {
		err = d.single(ctx, out)
	} else {
		err = d.segmented(ctx, out, pr.total, pr.etag, pr.lastmod)
	}
	if err != nil {
		return err
	}
	// --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(out, newSum, sumWant); err != nil {
			return err
		}
	}
	d.setStatus(download.Complete)
	return nil
}

// resolveName settles d.name and d.out after the probe. When -o is absent the
// name comes from Content-Disposition, else the final (post-redirect) URL, else
// 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 != "" {
			name = nameFromContentDisposition(cd)
		}
		if name == "" && pr.finalURL != nil {
			name = nameFromURLPath(pr.finalURL)
		}
		if name == "" {
			name = nameFromURL(d.primary())
		}
		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
	// was only needed for the report.
	if d.cfg.DryRun {
		return nil
	}

	// -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(out) {
		return nil
	}
	if fileExists(out) && !d.cfg.AllowOverwrite {
		if d.cfg.AutoRename {
			u, err := uniqueName(out)
			if err != nil {
				return err
			}
			d.setOut(u)
			d.setName(filepath.Base(u))
			return nil
		}
		return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", out, ErrOutputExists)
	}
	return nil
}

// retriesExhausted wraps the final cause once the retry budget is spent, so the
// caller — and the exit-code mapping in main — can still see whether it was DNS,
// a refused connection, or a timeout, rather than an opaque "too many retries".
func retriesExhausted(what string, tries int, cause error) error {
	if cause == nil {
		return fmt.Errorf("%s: failed", what)
	}
	// Only mention a count when we actually retried (tries >= 2). tries == 0
	// (unlimited) reaches this only after the loop is broken some other way and
	// has no meaningful fixed count, so it too falls through to the bare form.
	if tries > 1 {
		return fmt.Errorf("%s: %w (after %d tries)", what, cause, tries)
	}
	return fmt.Errorf("%s: %w", what, cause)
}

// withRetries runs attempt up to d.cfg.Tries times (0 = unlimited), the shared
// tries/fatal policy that probe, the single stream, and each segment all use so
// one transient DNS/connect/5xx does not fail the whole download (item [18]). It
// returns immediately on success (nil), on context cancellation, or on a fatal{}
// error (retrying cannot help), and wraps the final cause with retriesExhausted
// once the budget is spent so the underlying cause stays inspectable.
func (d *Download) withRetries(ctx context.Context, what string, attempt func() error) error {
	tries := d.cfg.Tries
	var lastErr error
	for n := 0; tries == 0 || n < tries; n++ {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		err := attempt()
		if err == nil || ctx.Err() != nil {
			return err
		}
		lastErr = err
		var fe fatal
		if errors.As(err, &fe) {
			return err // permanent: retrying cannot help
		}
	}
	return retriesExhausted(what, tries, lastErr)
}

// overMirrors calls fn against each mirror URL in turn, starting at offset start
// so callers can spread work across mirrors (segment index) and rotate on retry.
// It returns nil on the first success, ctx.Err() if cancelled between mirrors, or
// the last error once every mirror is spent.
func (d *Download) overMirrors(ctx context.Context, start int, fn func(uri string) error) error {
	var lastErr error
	for i := range d.uris {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if err := fn(d.mirror(start + i)); err != nil {
			lastErr = err
			continue
		}
		return nil
	}
	return lastErr
}

// probeRetry runs probe within the shared retry policy (item [18]). The probe
// result is captured through pr because withRetries only carries an error.
func (d *Download) probeRetry(ctx context.Context) (probeResult, error) {
	// Probe mirrors in order (primary first); the first that answers anchors the
	// size and validators for the whole download. A dead/404 primary falls over
	// to the next mirror.
	var pr probeResult
	err := d.overMirrors(ctx, 0, func(uri string) error {
		return d.withRetries(ctx, uri, func() error {
			var perr error
			pr, perr = d.probe(ctx, uri)
			return perr
		})
	})
	if err != nil {
		return probeResult{}, err
	}
	return pr, nil
}

// probe issues a one-byte ranged GET to learn the total size, whether the
// server honours ranges, and the validator headers used for safe resume. It
// also reports the final URL and Content-Disposition for name finalisation.
func (d *Download) probe(ctx context.Context, uri string) (probeResult, error) {
	req, err := d.request(ctx, uri, "bytes=0-0")
	if err != nil {
		return probeResult{}, err
	}
	resp, err := d.client.Do(req)
	if err != nil {
		return probeResult{}, err
	}
	defer resp.Body.Close()
	io.Copy(io.Discard, io.LimitReader(resp.Body, 1))

	pr := probeResult{
		etag:    resp.Header.Get("ETag"),
		lastmod: resp.Header.Get("Last-Modified"),
		cdisp:   resp.Header.Get("Content-Disposition"),
	}
	if resp.Request != nil {
		pr.finalURL = resp.Request.URL
	}

	switch resp.StatusCode {
	case http.StatusPartialContent:
		// Content-Range: bytes 0-0/12345
		pr.ranged = true
		if _, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok {
			pr.total = t
		}
		return pr, nil
	case http.StatusOK:
		pr.total = resp.ContentLength
		return pr, nil
	default:
		return probeResult{}, statusError(fmt.Sprintf("%s: %s", uri, resp.Status), resp.StatusCode)
	}
}

// segmented downloads total bytes over d.maxConns workers, with per-segment
// resume from the control file.
func (d *Download) segmented(ctx context.Context, out string, total int64, etag, lastmod string) error {
	segs := makeSegments(total, d.cfg.MinSplit, d.maxConns)
	if c := d.resumeControl(out, etag, lastmod); c != nil {
		segs = segsFromControl(c)
	} else if d.cfg.Continue && !fileExists(controlPath(out)) {
		// Resume a foreign/browser partial file: no .got sidecar exists, so treat
		// the bytes already on disk as a completed prefix and seed the segments
		// accordingly (item [2]): open the existing file and mark its bytes done.
		if fi, err := os.Stat(out); err == nil {
			seedSegmentsFromPrefix(segs, fi.Size())
		}
	}
	var have int64
	for i := range segs {
		have += segs[i].progress()
	}
	atomic.StoreInt64(&d.completed, have)

	if err := os.MkdirAll(filepath.Dir(out), 0o755); err != nil {
		return fmt.Errorf("create output dir: %w", err)
	}
	f, err := os.OpenFile(out, os.O_RDWR|os.O_CREATE, 0o644)
	if err != nil {
		return err
	}
	defer f.Close()
	if err := allocate(f, d.cfg.FileAlloc, total); err != nil {
		return err
	}
	// Force the file to exactly total bytes: prealloc/falloc only grow, so an
	// existing larger file (overwrite, or a changed resume target) would keep a
	// stale tail past the real content otherwise.
	if err := f.Truncate(total); err != nil {
		return err
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// Periodically persist progress so a crash can resume.
	saveDone := make(chan struct{})
	go d.saveLoop(ctx, out, total, etag, lastmod, segs, saveDone)

	// One worker per segment. Segments are non-overlapping byte ranges written
	// with WriteAt, so the workers share no cursor and need no coordination — a
	// finished worker simply exits. The division (makeSegments / a resumed
	// control) fixes the parallelism up front; there is no rebalancing.
	var (
		wg      sync.WaitGroup
		errOnce sync.Once
		runErr  error
	)
	for i := range segs {
		wg.Add(1)
		go func(s *seg) {
			defer wg.Done()
			if err := d.fetchSeg(ctx, f, s, total); err != nil {
				errOnce.Do(func() { runErr = err; cancel() })
			}
		}(&segs[i])
	}
	wg.Wait()
	cancel()
	<-saveDone

	if runErr != nil {
		return runErr // keep the control file for a later -c
	}
	snapshot(d.primary(), total, etag, lastmod, segs).save(out)
	removeControl(out)
	return nil
}

// fetchSeg downloads one segment, retrying from its resume point on error.
// Each attempt re-reads s.offset(), so a retry continues from the bytes already
// written rather than restarting the range.
func (d *Download) fetchSeg(ctx context.Context, f *os.File, s *seg, total int64) error {
	// Try each mirror in turn for this segment: a transient error retries the
	// same mirror under withRetries; an exhausted budget or a per-mirror permanent
	// error (a 404, a non-206, a length mismatch) falls over to the next mirror.
	// The segment fails only once every mirror is spent. Starting at s.index
	// spreads the segments across mirrors round-robin, and each attempt resumes
	// from s.offset().
	return d.overMirrors(ctx, s.index, func(uri string) error {
		return d.withRetries(ctx, fmt.Sprintf("segment %d", s.index), func() error {
			if s.done() {
				return nil
			}
			return d.fetchOnce(ctx, f, s, uri, total)
		})
	})
}

func (d *Download) fetchOnce(ctx context.Context, f *os.File, s *seg, uri string, total int64) error {
	reqCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.end))
	if err != nil {
		return err
	}
	atomic.AddInt32(&d.conns, 1)
	resp, err := d.client.Do(req)
	if err != nil {
		atomic.AddInt32(&d.conns, -1)
		return err
	}
	defer func() {
		resp.Body.Close()
		atomic.AddInt32(&d.conns, -1)
	}()
	if resp.StatusCode != http.StatusPartialContent {
		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. 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, s, body)
}

// ErrTimeout marks a transfer that stalled past the idle window (--timeout): a
// distinct "timed out" instead of the bare "context canceled" the cancelled
// request would otherwise surface (item [39]). main maps it to exit code 2 via
// errors.Is, so it must stay on the error chain (CONTRACT).
var ErrTimeout = errors.New("idle timeout: no data received")

// 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
	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) {
	n, err := ir.r.Read(p)
	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)", ErrTimeout, ir.idle)
	}
	return n, err
}

// 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.last.Store(time.Now().UnixNano())
	stop := make(chan struct{})
	go ir.watch(stop, cancel)
	return ir, func() { close(stop) }
}

// fatal marks an error that retrying cannot fix. Unwrap exposes the wrapped
// error so errors.Is/As traverse a fatal{} — main's exit-code mapping relies on
// errors.Is(err, ErrNotFound) seeing through it (CONTRACT).
type fatal struct{ error }

func (f fatal) Unwrap() error { return f.error }

// permanent reports whether an HTTP status is a client error that will not
// change on retry (4xx except the transient 408 Request Timeout and 429 Too
// Many Requests).
func permanent(code int) bool {
	return code >= 400 && code < 500 && code != 408 && code != 429
}

// statusError turns an unexpected HTTP status into the error the retry policy
// expects: a 404 becomes fatal{} wrapping ErrNotFound so it is permanent and
// main's exit-code mapping can still see it via errors.Is(err, ErrNotFound)
// (CONTRACT); any other permanent() 4xx becomes a plain fatal{} (retrying
// cannot help); everything else is returned bare so withRetries retries it.
// ctxMsg is the caller's "<what>: <status>" message (already formatted).
func statusError(ctxMsg string, code int) error {
	err := errors.New(ctxMsg)
	if code == http.StatusNotFound {
		return fatal{fmt.Errorf("%w: %s", ErrNotFound, err)}
	}
	if permanent(code) {
		return fatal{err}
	}
	return err
}

// pump copies the response body into the file at the segment's running offset,
// stopping at the segment end and respecting rate limits. Only this worker writes
// s.written, so the advance is a plain atomic add (Stat and the snapshot read it
// atomically); the loop ends when the range is filled.
func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader) error {
	buf := make([]byte, readBuf)
	for s.remaining() > 0 {
		n := int64(len(buf))
		if s.remaining() < n {
			n = s.remaining()
		}
		rd, err := body.Read(buf[:n])
		if rd > 0 {
			if _, werr := f.WriteAt(buf[:rd], s.offset()); werr != nil {
				return werr
			}
			s.addWritten(int64(rd))
			atomic.AddInt64(&d.completed, int64(rd))
			d.throttle(ctx, rd)
		}
		if err == io.EOF {
			// A clean EOF before the range is filled means the server sent a
			// short body (e.g. a re-chunking proxy). Treat it as an error so
			// the segment is retried from the advanced offset, never silently
			// left with a hole.
			if s.remaining() > 0 {
				return io.ErrUnexpectedEOF
			}
			return nil
		}
		if err != nil {
			return err
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
	}
	return nil
}

// single streams the whole resource over one connection (server has no ranges
// or the length is unknown). It runs through the shared retry policy so a
// transient failure mid-stream is retried instead of failing the download
// (item [1]); each attempt re-derives the resume offset from the on-disk file
// size, so a retry continues rather than restarting. With -c and an existing
// foreign partial file the first attempt resumes from that prefix (item [2]).
func (d *Download) single(ctx context.Context, out string) error {
	// foreignResume is honoured only on the first attempt: it decides whether the
	// bytes already on disk are a real prefix to keep (-c, no .got control) or a
	// stale file to truncate. On a retry the bytes on disk are whatever this run
	// wrote, so we always resume from them regardless.
	foreignResume := d.cfg.Continue && !fileExists(controlPath(out))
	first := true
	// One connection at a time, but try each mirror: a mirror that fails its
	// retry budget falls over to the next (resuming from what's already on disk).
	return d.overMirrors(ctx, 0, func(uri string) error {
		return d.withRetries(ctx, uri, func() error {
			resumeFromDisk := !first || foreignResume
			first = false
			return d.singleOnce(ctx, out, uri, resumeFromDisk)
		})
	})
}

// singleOnce performs one single-stream transfer attempt. When resumeFromDisk is
// set, the current on-disk file size is the resume point and the file is opened
// without O_TRUNC so the existing prefix survives; otherwise the file is
// truncated and the transfer starts from zero.
func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDisk bool) error {
	var resumeAt int64
	if resumeFromDisk {
		if fi, err := os.Stat(out); err == nil {
			resumeAt = fi.Size()
		}
	}

	rangeHdr := ""
	if resumeAt > 0 {
		rangeHdr = fmt.Sprintf("bytes=%d-", resumeAt)
	}

	reqCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	req, err := d.request(reqCtx, uri, rangeHdr)
	if err != nil {
		return err
	}
	atomic.AddInt32(&d.conns, 1)
	defer atomic.AddInt32(&d.conns, -1)
	resp, err := d.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	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 {
		resumeAt = 0
	}
	if resp.ContentLength > 0 {
		atomic.StoreInt64(&d.total, resumeAt+resp.ContentLength)
	}

	// Open without O_TRUNC when resuming so the existing prefix survives; the
	// fresh path truncates to drop any stale tail.
	flags := os.O_RDWR | os.O_CREATE
	if resumeAt == 0 {
		flags |= os.O_TRUNC
	}
	if err := os.MkdirAll(filepath.Dir(out), 0o755); err != nil {
		return fmt.Errorf("create output dir: %w", err)
	}
	f, err := os.OpenFile(out, flags, 0o644)
	if err != nil {
		return err
	}
	defer f.Close()

	body, stop := d.idleGuard(resp.Body, cancel)
	defer stop()
	buf := make([]byte, readBuf)
	off := resumeAt
	atomic.StoreInt64(&d.completed, off)
	// When the length is known, track bytes received in this stream so a body
	// that ends early (a proxy that closes mid-transfer) is reported as a
	// retryable short read instead of a false Complete (item [1], mirrors pump).
	var got int64
	want := resp.ContentLength // -1 when unknown
	for {
		rd, err := body.Read(buf)
		if rd > 0 {
			if _, werr := f.WriteAt(buf[:rd], off); werr != nil {
				return werr
			}
			off += int64(rd)
			got += int64(rd)
			atomic.AddInt64(&d.completed, int64(rd)) // add deltas, as pump does
			d.throttle(ctx, rd)
		}
		if err == io.EOF {
			if want >= 0 && got < want {
				return io.ErrUnexpectedEOF // short body: retry from the resumed offset
			}
			removeControl(out) // drop any stale control file from a prior segmented run
			return nil
		}
		if err != nil {
			return err
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
	}
}

// saveInterval is how often the .got control file is rewritten so a crash can
// resume; on a clean finish it is removed.
const saveInterval = 60 * time.Second

func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, segs []seg, done chan<- struct{}) {
	defer close(done)
	t := time.NewTicker(saveInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			snapshot(d.primary(), total, etag, lastmod, segs).save(out)
			return
		case <-t.C:
			snapshot(d.primary(), total, etag, lastmod, segs).save(out)
		}
	}
}

func (d *Download) resumeControl(out, etag, lastmod string) *control {
	if !d.cfg.Continue {
		return nil
	}
	return loadControl(out, d.primary(), atomic.LoadInt64(&d.total), etag, lastmod)
}

func (d *Download) throttle(ctx context.Context, n int) {
	for _, l := range d.limit {
		_ = l.WaitN(ctx, n)
	}
}

func (d *Download) request(ctx context.Context, uri, rangeHdr string) (*http.Request, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, uri, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", userAgent)
	// --header sets arbitrary request headers, applied last so they can override
	// the User-Agent (or add a Referer, Authorization, Cookie, ...) when needed.
	for _, h := range d.cfg.Headers {
		if k, v, ok := strings.Cut(h, ":"); ok {
			req.Header.Set(strings.TrimSpace(k), strings.TrimSpace(v))
		}
	}
	if rangeHdr != "" {
		req.Header.Set("Range", rangeHdr)
	}
	return req, nil
}

// seedSegmentsFromPrefix marks the first prefix bytes of the file as already
// downloaded, segment by segment. A foreign partial file has no per-segment
// metadata, so the only thing we can trust is that bytes [0,prefix) are present;
// we fill whole leading segments and a partial one at the boundary.
func seedSegmentsFromPrefix(segs []seg, prefix int64) {
	for i := range segs {
		if prefix <= segs[i].start {
			break
		}
		fill := prefix - segs[i].start
		if fill > segs[i].length() {
			fill = segs[i].length()
		}
		segs[i].written = fill
	}
}

func segsFromControl(c *control) []seg {
	segs := make([]seg, len(c.Segs))
	for i, st := range c.Segs {
		segs[i] = seg{index: i, start: st.Start, end: st.End, written: st.Written}
	}
	return segs
}

func nameFromURL(u string) string {
	parsed, err := url.Parse(u)
	if err != nil {
		return "index"
	}
	return nameFromURLPath(parsed)
}

// nameFromURLPath takes the basename of a URL's path, percent-decoded, falling
// back to "index" for a directory-style URL.
func nameFromURLPath(u *url.URL) string {
	base := path.Base(u.Path)
	if dec, err := url.PathUnescape(base); err == nil {
		base = dec
	}
	if base == "" || base == "." || base == "/" {
		return "index"
	}
	return base
}

// nameFromContentDisposition extracts the filename from a Content-Disposition
// header per RFC 6266, preferring filename* when present (mime.ParseMediaType
// already decodes it). Returns "" when there is no usable filename, so the
// caller can fall back to the URL. Path separators are stripped to keep the
// server from writing outside the output directory.
func nameFromContentDisposition(cd string) string {
	_, params, err := mime.ParseMediaType(cd)
	if err != nil {
		return ""
	}
	name := params["filename"]
	if name == "" {
		return ""
	}
	name = filepath.Base(filepath.FromSlash(name))
	if name == "" || name == "." || name == string(filepath.Separator) {
		return ""
	}
	return name
}

// uniqueName returns the first non-colliding variant of p by inserting ".N"
// before the final extension (stem.N.ext), or "base.N" when there is no
// extension. It caps auto file renaming at 9999; a
// candidate that exists is acceptable when its .got control file also exists
// (it is a resumable partial). Past the cap it returns an error (item [13]).
func uniqueName(p string) (string, error) {
	stem, ext := splitExt(p)
	for i := 1; i < 10000; i++ {
		cand := fmt.Sprintf("%s.%d%s", stem, i, ext)
		if !fileExists(cand) || fileExists(controlPath(cand)) {
			return cand, nil
		}
	}
	return "", fmt.Errorf("auto file renaming failed for %s: too many existing files", p)
}

// splitExt splits p into stem and extension: a leading dot is
// not an extension (".bashrc" stays whole), and the extension search ignores
// any directory component.
func splitExt(p string) (stem, ext string) {
	ext = filepath.Ext(p)
	if ext == "" {
		return p, ""
	}
	base := filepath.Base(p)
	// A dotfile with no other dot ("/dir/.ext") has no extension.
	if strings.HasPrefix(base, ".") && strings.Count(base, ".") == 1 {
		return p, ""
	}
	return p[:len(p)-len(ext)], ext
}

func fileExists(p string) bool {
	_, err := os.Stat(p)
	return err == nil
}

// parseContentRange parses "bytes 0-0/12345" into start, end, total.
func parseContentRange(s string) (start, end, total int64, ok bool) {
	s = strings.TrimPrefix(strings.TrimSpace(s), "bytes ")
	rng, totalStr, found := strings.Cut(s, "/")
	if !found {
		return 0, 0, 0, false
	}
	startStr, endStr, found := strings.Cut(rng, "-")
	if !found {
		return 0, 0, 0, false
	}
	start, e1 := strconv.ParseInt(startStr, 10, 64)
	end, e2 := strconv.ParseInt(endStr, 10, 64)
	total, e3 := strconv.ParseInt(totalStr, 10, 64)
	if e1 != nil || e2 != nil || e3 != nil {
		return 0, 0, 0, false
	}
	return start, end, total, true
}