// 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 "=" 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 ": " 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 }