// 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" "crypto/tls" "crypto/x509" "errors" "fmt" "hash" "io" "mime" "net" "net/http" "net/http/cookiejar" "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 // 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") // ErrTooSlow reports a transfer aborted by --lowest-speed-limit. main maps it to // exit code 5 via errors.Is, so it must stay on the error chain (CONTRACT). var ErrTooSlow = errors.New("download speed too low") // 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 MaxConnPerServer int // --max-connection-per-server: per-host connection cap MinSplit int64 Tries int // 0 = unlimited Timeout time.Duration FileAlloc string // none | prealloc | trunc | falloc Continue bool AllowOverwrite bool AutoRename bool Headers []string UserAgent string Referer string Proxy string CheckCert bool Limit int64 // per-download bytes/s, 0 = unlimited OverallLimiter *rate.Limiter // shared across all downloads, may be nil // RetryWait is the pause between retries of a transient failure // (--retry-wait). 0 means retry immediately with no sleep. RetryWait time.Duration // AutoSaveInterval controls how often the .got control file is rewritten // (--auto-save-interval). <=0 falls back to 60s. AutoSaveInterval time.Duration // ConnectTimeout is the time allowed to establish the TCP/TLS connection // (--connect-timeout, default 60s), distinct from Timeout which // covers stalls once data is flowing. <=0 falls back to Timeout. ConnectTimeout time.Duration // LoadCookies / SaveCookies are paths to a Netscape cookies.txt read into // the client's jar before the download and written back on exit. LoadCookies string SaveCookies string // ConditionalGet sends If-Modified-Since (from the existing local file's // mtime) so a 304 reports success without re-downloading // (--conditional-get). ConditionalGet bool // RemoteTime sets the finished file's mtime from the server's Last-Modified // header (-R / --remote-time). RemoteTime bool // HTTPUser / HTTPPasswd enable HTTP Basic auth (--http-user/passwd). HTTPUser string HTTPPasswd string // LowestSpeedLimit aborts a download whose speed stays at or below this many // bytes/sec for a sustained window (--lowest-speed-limit). 0 = off. LowestSpeedLimit int64 // Checksum is a "=" 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 // CACert is a path to a PEM bundle of CA certificates used to verify HTTPS // servers (--ca-certificate). Empty uses the system roots. CACert string // DisableIPv6 forces IPv4-only dialing (--disable-ipv6). DisableIPv6 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)*max-connection-per-server) cfg Config client *http.Client limit []*rate.Limiter 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). 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 // SaveCookies can ask the jar for the cookies of each contacted host. mu sync.Mutex seenURLs []*url.URL // live counters, read by Stat from any goroutine total int64 // -1 until known completed int64 status int32 // download.Status conns int32 // active connections } // loadCAPool reads a PEM bundle of CA certificates for verifying HTTPS servers // (--ca-certificate). An unreadable file or one containing no // certificates is an error rather than a silent fall-back to the system roots. func loadCAPool(path string) (*x509.CertPool, error) { pem, err := os.ReadFile(path) if err != nil { return nil, err } pool := x509.NewCertPool() if !pool.AppendCertsFromPEM(pem) { return nil, fmt.Errorf("ca-certificate %s: no certificates found", path) } return pool, nil } // 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 { // --connect-timeout bounds only connection establishment; once data flows // --timeout (the idle guard) takes over. The dial and the TLS handshake each // get the full connectTimeout (Go applies it per phase), so an HTTPS connect // can take up to ~2x connectTimeout in the worst case — there is no single // shared connect budget. connectTimeout := cfg.ConnectTimeout if connectTimeout <= 0 { connectTimeout = cfg.Timeout } var initErr error // --ca-certificate: verify HTTPS servers against a custom CA bundle instead of // the system roots. A bad bundle becomes an initErr so Run fails loudly. tlsCfg := &tls.Config{InsecureSkipVerify: !cfg.CheckCert} if cfg.CACert != "" { if pool, err := loadCAPool(cfg.CACert); err != nil { initErr = err } else { tlsCfg.RootCAs = pool } } // --disable-ipv6: force IPv4 by dialing "tcp4"; a plain "tcp" would let the // resolver hand back a v6 address. dialer := &net.Dialer{Timeout: connectTimeout} dialContext := dialer.DialContext if cfg.DisableIPv6 { dialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { if network == "tcp" { network = "tcp4" } return dialer.DialContext(ctx, network, addr) } } // MaxConnsPerHost is the per-host cap (--max-connection-per-server): with // mirrors the total worker count can exceed it, but no single host does. perHost := cfg.MaxConnPerServer if perHost < 1 { perHost = 1 } tr := &http.Transport{ Proxy: http.ProxyFromEnvironment, MaxConnsPerHost: perHost, MaxIdleConnsPerHost: perHost, ResponseHeaderTimeout: cfg.Timeout, TLSHandshakeTimeout: connectTimeout, DialContext: dialContext, TLSClientConfig: tlsCfg, } if cfg.Proxy != "" { 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) } } // Seed the cookie jar from --load-cookies; a load error becomes an initErr so // Run surfaces it rather than silently downloading uncredentialed. var jar *cookiejar.Jar if cfg.LoadCookies != "" || cfg.SaveCookies != "" { if j, err := loadCookieJar(cfg.LoadCookies); err != nil { if initErr == nil { initErr = err } } else { jar = j } } var limit []*rate.Limiter if cfg.OverallLimiter != nil { limit = append(limit, cfg.OverallLimiter) } if cfg.Limit > 0 { limit = append(limit, rate.NewLimiter(rate.Limit(cfg.Limit), download.LimiterBurst(cfg.Limit))) } // 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: --split connections are spread across the mirrors, capped // at --max-connection-per-server per host, so the ceiling is // min(split, M*max-conn-per-server). One mirror reduces to min(split, x). maxConns := cfg.Split if m := max(1, len(uris)) * perHost; maxConns > m { maxConns = m } if maxConns < 1 { maxConns = 1 } // Only attach the jar when one was built: a typed-nil *cookiejar.Jar stored // in the http.Client.Jar interface is non-nil and would panic on use. client := &http.Client{Transport: tr} if jar != nil { client.Jar = jar } d := &Download{ uris: uris, maxConns: maxConns, cfg: cfg, client: client, limit: limit, jar: jar, 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 (used by --follow-torrent). 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 notModified bool // server answered 304 to a conditional GET } // 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 } // --conditional-get: the server said our local copy is current, so there is // nothing to download. Report success without touching the file. if pr.notModified { d.setStatus(download.Complete) return nil } 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 } // --lowest-speed-limit: cancel the transfer if its speed stays at or below // 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, out) } else { err = d.segmented(xferCtx, out, pr.total, pr.etag, pr.lastmod) } stopGuard() if err != nil { // 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 } // --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 } } // -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(out, t, t) } } 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 } // backoff waits RetryWait between attempts, but never after the final attempt, // and returns ctx.Err() if cancelled while waiting (item [20]). func backoff(ctx context.Context, wait time.Duration, tries, attempt int) error { if wait <= 0 || (tries != 0 && attempt >= tries-1) { return nil } select { case <-ctx.Done(): return ctx.Err() case <-time.After(wait): 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/backoff/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), sleeps RetryWait between attempts, and // wraps the final cause with retriesExhausted(what, ...) 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 } if err := backoff(ctx, d.cfg.RetryWait, tries, n); err != nil { return err } } 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 } // --conditional-get: ask the server to skip the transfer when our local file // 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 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)) } } 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 } if resp.StatusCode == http.StatusNotModified { pr.notModified = true return pr, nil } 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.endOff())) 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) } } // speedStartupGrace is how long a download may stay slow before the // lowest-speed-limit watchdog starts checking it: a startup idle time of // 10 seconds. const speedStartupGrace = 10 * time.Second // belowSpeed reports whether the bytes transferred between two cumulative // d.completed samples (prev -> now over one window) put the download at or // below limit. A negative delta is not a slow second: when a resumed Range is // answered by a 200 the transfer restarts from offset 0, so d.completed jumps // backward on that retry. Treating that backward jump as "below limit" would // false-abort a healthy download, so a reset (delta < 0) reports false; the // caller has already advanced its baseline, so the next full window measures the // real speed. Only a genuine non-negative delta at or under the limit aborts. func belowSpeed(prev, now, limit int64) bool { delta := now - prev if delta < 0 { return false // counter reset, not a slow window } return delta <= limit } // speedGuard implements --lowest-speed-limit. It returns a child context that // the watcher cancels (setting a flag) once the measured download speed has // stayed at or below the limit past the startup grace, plus a stop func and a // predicate reporting whether the watcher fired. When the limit is 0 it is a // no-op passthrough. // // Speed is sampled from d.completed over a one-second window: a single slow // second after the grace aborts the download as too slow. func (d *Download) speedGuard(ctx context.Context) (context.Context, func(), func() bool) { if d.cfg.LowestSpeedLimit <= 0 { return ctx, func() {}, func() bool { return false } } child, cancel := context.WithCancel(ctx) var fired atomic.Bool done := make(chan struct{}) go func() { defer close(done) start := time.Now() ticker := time.NewTicker(time.Second) defer ticker.Stop() last := atomic.LoadInt64(&d.completed) for { select { case <-child.Done(): return case <-ticker.C: now := atomic.LoadInt64(&d.completed) prev := last last = now if time.Since(start) < speedStartupGrace { continue } if belowSpeed(prev, now, d.cfg.LowestSpeedLimit) { fired.Store(true) cancel() return } } } }() stop := func() { cancel() <-done } return child, stop, func() bool { return fired.Load() } } // 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 ": " 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() } } } func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, segs []seg, done chan<- struct{}) { defer close(done) interval := d.cfg.AutoSaveInterval if interval <= 0 { interval = 60 * time.Second // --auto-save-interval default } t := time.NewTicker(interval) 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", d.cfg.UserAgent) if d.cfg.Referer != "" { req.Header.Set("Referer", d.cfg.Referer) } if d.cfg.HTTPUser != "" { req.SetBasicAuth(d.cfg.HTTPUser, d.cfg.HTTPPasswd) } 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) } if d.jar != nil { d.recordURL(req.URL) } return req, nil } // recordURL remembers a contacted URL so SaveCookies can later read back the // jar's cookies for that host. Duplicates are tolerated; saveCookieJar dedups. func (d *Download) recordURL(u *url.URL) { d.mu.Lock() d.seenURLs = append(d.seenURLs, u) d.mu.Unlock() } // SaveCookies writes the client's cookie jar back to the --save-cookies file, // if one was configured. It is a no-op when cookies are not in use. main calls // it once per HTTP download on exit (cookies are saved at session end). func (d *Download) SaveCookies() error { if d.cfg.SaveCookies == "" || d.jar == nil { return nil } d.mu.Lock() urls := append([]*url.URL{}, d.seenURLs...) d.mu.Unlock() return saveCookieJar(d.jar, urls, d.cfg.SaveCookies) } // 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 }