httpdl: fix progress-reporter race on name/out; harden mirror & resume validation
Findings from a Rob-Pike-lens review (bugs/races/network), each verified against the code before fixing: - httpdl: name/out are now atomic.Pointer[string] -- the engine publishes a download to the reporter before Run resolves the name, so Stat raced the write (bt already did this) - httpdl: a malformed --proxy fails loudly instead of silently bypassing it - httpdl: a 206 must carry a matching Content-Range; a 200 in segmented mode is fatal so it fails over instead of burning the retry budget - httpdl: single-stream mirror failover validates the range before appending; ErrTooSlow only when the error is a real ctx cancellation - httpdl: idle guard tracks progress by timestamp (no Reset/Stop race, no sticky fired flag) - httpdl/control: reject a resume file whose segments don't tile [0,total) - bt: clamp the listen-port range; verify on-disk data before choosing pieces under --check-integrity - cli: reject size overflow; show --seed-time=MIN; clamp --select-file range - progress/engine/main: clamp ETA against int64 overflow; show queued downloads as waiting; join the reporter on exit instead of a 20ms sleep
This commit is contained in:
22
bt/bt.go
22
bt/bt.go
@@ -68,10 +68,17 @@ func ParsePorts(spec string) []int {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
for i := a; i <= b; i++ {
|
// Clamp to the valid port space before iterating: a parseable but oversized
|
||||||
if i >= 1 && i <= 65535 {
|
// spec like "1-9999999999" would otherwise spin for ~10^10 iterations and
|
||||||
ports = append(ports, i)
|
// hang startup, even though only [1,65535] can ever be appended.
|
||||||
|
if a < 1 {
|
||||||
|
a = 1
|
||||||
}
|
}
|
||||||
|
if b > 65535 {
|
||||||
|
b = 65535
|
||||||
|
}
|
||||||
|
for i := a; i <= b; i++ {
|
||||||
|
ports = append(ports, i)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return ports
|
return ports
|
||||||
@@ -306,14 +313,17 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := d.choose(t); err != nil {
|
// --check-integrity: re-hash the existing on-disk data BEFORE arming the
|
||||||
return err
|
// request loop, so already-good pieces are not re-requested from peers (aria2
|
||||||
}
|
// verifies first, then fetches only what is missing).
|
||||||
if d.opts.CheckIntegrity {
|
if d.opts.CheckIntegrity {
|
||||||
if err := t.VerifyDataContext(ctx); err != nil {
|
if err := t.VerifyDataContext(ctx); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if err := d.choose(t); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
if err := d.wait(ctx, t); err != nil {
|
if err := d.wait(ctx, t); err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
@@ -97,7 +97,12 @@ func flagLabel(o *Opt) string {
|
|||||||
case Int:
|
case Int:
|
||||||
b.WriteString("=N")
|
b.WriteString("=N")
|
||||||
case Float:
|
case Float:
|
||||||
|
// seed-time is a count of minutes; only seed-ratio is an actual ratio.
|
||||||
|
if o.Long == "seed-time" {
|
||||||
|
b.WriteString("=MIN")
|
||||||
|
} else {
|
||||||
b.WriteString("=RATIO")
|
b.WriteString("=RATIO")
|
||||||
|
}
|
||||||
case Enum:
|
case Enum:
|
||||||
b.WriteString("=" + strings.Join(o.Enum, "|"))
|
b.WriteString("=" + strings.Join(o.Enum, "|"))
|
||||||
default:
|
default:
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package cli
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
)
|
)
|
||||||
@@ -105,5 +106,10 @@ func parseSize(s string) (int64, error) {
|
|||||||
if n < 0 {
|
if n < 0 {
|
||||||
return 0, fmt.Errorf("negative size %q", s)
|
return 0, fmt.Errorf("negative size %q", s)
|
||||||
}
|
}
|
||||||
|
// Reject a value whose unit multiply would overflow int64 and silently wrap to
|
||||||
|
// a bogus (positive or negative) byte count.
|
||||||
|
if mult > 1 && n > math.MaxInt64/mult {
|
||||||
|
return 0, fmt.Errorf("size %q too large", s)
|
||||||
|
}
|
||||||
return n * mult, nil
|
return n * mult, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -46,6 +46,12 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
|||||||
go func(i int, d Download) {
|
go func(i int, d Download) {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
|
|
||||||
|
// Track immediately so a download still queued for a concurrency slot
|
||||||
|
// is visible to the progress reporter as Waiting, rather than hidden
|
||||||
|
// until it starts running.
|
||||||
|
e.track(d)
|
||||||
|
defer e.untrack(d)
|
||||||
|
|
||||||
// Acquire a concurrency slot, or bail if we're shutting down
|
// Acquire a concurrency slot, or bail if we're shutting down
|
||||||
// before this download ever started.
|
// before this download ever started.
|
||||||
select {
|
select {
|
||||||
@@ -56,9 +62,6 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
|||||||
}
|
}
|
||||||
defer func() { <-slots }()
|
defer func() { <-slots }()
|
||||||
|
|
||||||
e.track(d)
|
|
||||||
defer e.untrack(d)
|
|
||||||
|
|
||||||
err := d.Run(ctx)
|
err := d.Run(ctx)
|
||||||
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
|
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
|
||||||
}(i, d)
|
}(i, d)
|
||||||
@@ -74,7 +77,8 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// Snapshot returns a Stat for every currently running download.
|
// Snapshot returns a Stat for every tracked download — those running plus those
|
||||||
|
// still queued for a concurrency slot (reported as Waiting).
|
||||||
func (e *Engine) Snapshot() []Stat {
|
func (e *Engine) Snapshot() []Stat {
|
||||||
e.mu.Lock()
|
e.mu.Lock()
|
||||||
defer e.mu.Unlock()
|
defer e.mu.Unlock()
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"sort"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -75,7 +76,35 @@ func loadControl(out, url string, total int64, etag, lastmod string) *control {
|
|||||||
if (lastmod != "" || c.LastModified != "") && c.LastModified != lastmod {
|
if (lastmod != "" || c.LastModified != "") && c.LastModified != lastmod {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
// Reject a control file whose segments do not exactly tile [0,total): a
|
||||||
|
// truncated/corrupted/hand-edited sidecar that still parses as JSON could
|
||||||
|
// otherwise mark a segment done() without its bytes on disk (inflated Written)
|
||||||
|
// or leave an un-downloaded hole, both of which would be reported as a complete
|
||||||
|
// file. We restart cleanly instead of trusting it.
|
||||||
|
if !validSegs(c.Segs, c.Total) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
return &c
|
return &c
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// validSegs reports whether segs cover [0,total) with no gap or overlap and a
|
||||||
|
// sane written count for each. The recorded order is not sorted (a steal appends
|
||||||
|
// a tail), so we check coverage on a sorted copy.
|
||||||
|
func validSegs(segs []segState, total int64) bool {
|
||||||
|
if total <= 0 || len(segs) == 0 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
sorted := append([]segState(nil), segs...)
|
||||||
|
sort.Slice(sorted, func(i, j int) bool { return sorted[i].Start < sorted[j].Start })
|
||||||
|
var next int64
|
||||||
|
for _, s := range sorted {
|
||||||
|
length := s.End - s.Start + 1
|
||||||
|
if s.Start != next || s.End < s.Start || s.Written < 0 || s.Written > length {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
next = s.End + 1
|
||||||
|
}
|
||||||
|
return next == total
|
||||||
|
}
|
||||||
|
|
||||||
func removeControl(out string) { os.Remove(controlPath(out)) }
|
func removeControl(out string) { os.Remove(controlPath(out)) }
|
||||||
|
|||||||
196
httpdl/httpdl.go
196
httpdl/httpdl.go
@@ -127,12 +127,13 @@ type Download struct {
|
|||||||
jar *cookiejar.Jar // non-nil when cookies are in use, for save-cookies
|
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
|
// name and out are resolved in Run after the probe response is known (the
|
||||||
// final post-redirect URL and any Content-Disposition can change them), then
|
// final post-redirect URL and any Content-Disposition can change them). The
|
||||||
// never mutated again, so Stat can read name from another goroutine without a
|
// engine publishes a Download to the progress reporter before Run starts, so
|
||||||
// lock once Run has set it. They are seeded in New so a Stat before Run still
|
// Stat reads these from the reporter goroutine while resolveName writes them;
|
||||||
// has a sensible best-effort name.
|
// they are atomic.Pointer[string] (exactly like bt.Download.name) to keep that
|
||||||
name string
|
// read/write race-free. Seeded in New so a Stat before Run has a best-effort name.
|
||||||
out string
|
name atomic.Pointer[string]
|
||||||
|
out atomic.Pointer[string]
|
||||||
initErr error
|
initErr error
|
||||||
|
|
||||||
// mu guards seenURLs, which records every URL we sent a request to so
|
// mu guards seenURLs, which records every URL we sent a request to so
|
||||||
@@ -232,7 +233,21 @@ func New(uris []string, cfg Config) *Download {
|
|||||||
TLSClientConfig: tlsCfg,
|
TLSClientConfig: tlsCfg,
|
||||||
}
|
}
|
||||||
if cfg.Proxy != "" {
|
if cfg.Proxy != "" {
|
||||||
if pu, err := url.Parse(cfg.Proxy); err == nil {
|
pu, err := url.Parse(cfg.Proxy)
|
||||||
|
switch {
|
||||||
|
case err != nil:
|
||||||
|
// A malformed proxy must fail loudly, not silently fall back to the
|
||||||
|
// environment proxy (or a direct connection) and leak the real IP.
|
||||||
|
if initErr == nil {
|
||||||
|
initErr = fmt.Errorf("proxy %q: %w", cfg.Proxy, err)
|
||||||
|
}
|
||||||
|
case pu.Host == "":
|
||||||
|
// "host:port" with no scheme parses to an opaque URL with no Host, which
|
||||||
|
// http.ProxyURL would route nowhere; require an explicit scheme.
|
||||||
|
if initErr == nil {
|
||||||
|
initErr = fmt.Errorf("proxy %q: missing scheme (use scheme://host:port)", cfg.Proxy)
|
||||||
|
}
|
||||||
|
default:
|
||||||
tr.Proxy = http.ProxyURL(pu)
|
tr.Proxy = http.ProxyURL(pu)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -294,29 +309,47 @@ func New(uris []string, cfg Config) *Download {
|
|||||||
client.Jar = jar
|
client.Jar = jar
|
||||||
}
|
}
|
||||||
|
|
||||||
return &Download{
|
d := &Download{
|
||||||
uris: uris,
|
uris: uris,
|
||||||
maxConns: maxConns,
|
maxConns: maxConns,
|
||||||
cfg: cfg,
|
cfg: cfg,
|
||||||
client: client,
|
client: client,
|
||||||
limit: limit,
|
limit: limit,
|
||||||
jar: jar,
|
jar: jar,
|
||||||
name: name,
|
|
||||||
out: out,
|
|
||||||
total: -1,
|
total: -1,
|
||||||
initErr: initErr,
|
initErr: initErr,
|
||||||
}
|
}
|
||||||
|
d.setName(name)
|
||||||
|
d.setOut(out)
|
||||||
|
return d
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Download) Name() string { return d.name }
|
func (d *Download) Name() string { return d.loadName() }
|
||||||
|
|
||||||
// Path returns the resolved output file path (used by --follow-torrent).
|
// Path returns the resolved output file path (used by --follow-torrent).
|
||||||
func (d *Download) Path() string { return d.out }
|
func (d *Download) Path() string { return d.loadOut() }
|
||||||
|
|
||||||
|
// name and out are read by Stat from the reporter goroutine while Run resolves
|
||||||
|
// them, so they are stored atomically (mirroring bt.Download.name).
|
||||||
|
func (d *Download) setName(s string) { d.name.Store(&s) }
|
||||||
|
func (d *Download) setOut(s string) { d.out.Store(&s) }
|
||||||
|
func (d *Download) loadName() string {
|
||||||
|
if p := d.name.Load(); p != nil {
|
||||||
|
return *p
|
||||||
|
}
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
func (d *Download) loadOut() string {
|
||||||
|
if p := d.out.Load(); p != nil {
|
||||||
|
return *p
|
||||||
|
}
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
||||||
func (d *Download) Stat() download.Stat {
|
func (d *Download) Stat() download.Stat {
|
||||||
return download.Stat{
|
return download.Stat{
|
||||||
Name: d.name,
|
Name: d.loadName(),
|
||||||
ID: d.out, // resolved output path; stable per process (CONTRACT)
|
ID: d.loadOut(), // resolved output path; stable per process (CONTRACT)
|
||||||
IsBT: false,
|
IsBT: false,
|
||||||
Status: download.Status(atomic.LoadInt32(&d.status)),
|
Status: download.Status(atomic.LoadInt32(&d.status)),
|
||||||
Total: atomic.LoadInt64(&d.total),
|
Total: atomic.LoadInt64(&d.total),
|
||||||
@@ -400,15 +433,19 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
|||||||
// the limit past a startup grace. The watcher cancels xferCtx and sets a
|
// 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
|
// flag so we can return the distinct "too slow" error rather than a bare
|
||||||
// cancellation.
|
// cancellation.
|
||||||
|
out := d.loadOut()
|
||||||
xferCtx, stopGuard, tooSlow := d.speedGuard(ctx)
|
xferCtx, stopGuard, tooSlow := d.speedGuard(ctx)
|
||||||
if !pr.ranged || pr.total <= 0 {
|
if !pr.ranged || pr.total <= 0 {
|
||||||
err = d.single(xferCtx, d.out)
|
err = d.single(xferCtx, out)
|
||||||
} else {
|
} else {
|
||||||
err = d.segmented(xferCtx, d.out, pr.total, pr.etag, pr.lastmod)
|
err = d.segmented(xferCtx, out, pr.total, pr.etag, pr.lastmod)
|
||||||
}
|
}
|
||||||
stopGuard()
|
stopGuard()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if tooSlow() {
|
// Only translate to ErrTooSlow when the transfer actually ended in the
|
||||||
|
// guard's cancellation; a write error or mirror exhaustion that merely
|
||||||
|
// raced the guard firing keeps its own (real) cause and exit code.
|
||||||
|
if tooSlow() && errors.Is(err, context.Canceled) {
|
||||||
return fmt.Errorf("%s: %w (<= %d bytes/sec)", d.primary(), ErrTooSlow, d.cfg.LowestSpeedLimit)
|
return fmt.Errorf("%s: %w (<= %d bytes/sec)", d.primary(), ErrTooSlow, d.cfg.LowestSpeedLimit)
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
@@ -416,14 +453,14 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
|||||||
// --checksum: verify the finished file before declaring success, so a
|
// --checksum: verify the finished file before declaring success, so a
|
||||||
// corrupted or tampered download fails (--checksum, exit code 32).
|
// corrupted or tampered download fails (--checksum, exit code 32).
|
||||||
if newSum != nil {
|
if newSum != nil {
|
||||||
if err = verifyFile(d.out, newSum, sumWant); err != nil {
|
if err = verifyFile(out, newSum, sumWant); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// -R / --remote-time: stamp the finished file with the server's mtime.
|
// -R / --remote-time: stamp the finished file with the server's mtime.
|
||||||
if d.cfg.RemoteTime && pr.lastmod != "" {
|
if d.cfg.RemoteTime && pr.lastmod != "" {
|
||||||
if t, err := http.ParseTime(pr.lastmod); err == nil {
|
if t, err := http.ParseTime(pr.lastmod); err == nil {
|
||||||
os.Chtimes(d.out, t, t)
|
os.Chtimes(out, t, t)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
d.setStatus(download.Complete)
|
d.setStatus(download.Complete)
|
||||||
@@ -435,6 +472,7 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
|||||||
// the original URL. Overwrite / auto-rename is then resolved against the final
|
// 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]).
|
// path, honouring -c so a resumable file is never renamed (items [2],[8],[13]).
|
||||||
func (d *Download) resolveName(pr probeResult) error {
|
func (d *Download) resolveName(pr probeResult) error {
|
||||||
|
out := d.loadOut()
|
||||||
if d.cfg.Out == "" {
|
if d.cfg.Out == "" {
|
||||||
name := ""
|
name := ""
|
||||||
if cd := pr.cdisp; cd != "" {
|
if cd := pr.cdisp; cd != "" {
|
||||||
@@ -446,8 +484,9 @@ func (d *Download) resolveName(pr probeResult) error {
|
|||||||
if name == "" {
|
if name == "" {
|
||||||
name = nameFromURL(d.primary())
|
name = nameFromURL(d.primary())
|
||||||
}
|
}
|
||||||
d.name = name
|
out = filepath.Join(d.cfg.Dir, name)
|
||||||
d.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
|
// --dry-run writes nothing, so skip overwrite/auto-rename resolution; the name
|
||||||
@@ -458,20 +497,20 @@ func (d *Download) resolveName(pr probeResult) error {
|
|||||||
|
|
||||||
// -c resumes either our own .got sidecar or a foreign/browser partial file,
|
// -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]).
|
// so an existing file is never auto-renamed under -c (item [2]).
|
||||||
if d.cfg.Continue && fileExists(d.out) {
|
if d.cfg.Continue && fileExists(out) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
if fileExists(d.out) && !d.cfg.AllowOverwrite {
|
if fileExists(out) && !d.cfg.AllowOverwrite {
|
||||||
if d.cfg.AutoRename {
|
if d.cfg.AutoRename {
|
||||||
u, err := uniqueName(d.out)
|
u, err := uniqueName(out)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
d.out = u
|
d.setOut(u)
|
||||||
d.name = filepath.Base(u)
|
d.setName(filepath.Base(u))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", d.out, ErrOutputExists)
|
return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", out, ErrOutputExists)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -587,8 +626,8 @@ func (d *Download) probe(ctx context.Context, uri string) (probeResult, error) {
|
|||||||
// is already current. Only done when there is no .got control file (an
|
// 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
|
// in-progress resume must not be short-circuited), using the local file's
|
||||||
// mtime for If-Modified-Since.
|
// mtime for If-Modified-Since.
|
||||||
if d.cfg.ConditionalGet && !fileExists(controlPath(d.out)) {
|
if out := d.loadOut(); d.cfg.ConditionalGet && !fileExists(controlPath(out)) {
|
||||||
if fi, err := os.Stat(d.out); err == nil {
|
if fi, err := os.Stat(out); err == nil {
|
||||||
req.Header.Set("If-Modified-Since", fi.ModTime().UTC().Format(http.TimeFormat))
|
req.Header.Set("If-Modified-Since", fi.ModTime().UTC().Format(http.TimeFormat))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -757,60 +796,94 @@ func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, u
|
|||||||
atomic.AddInt32(&d.conns, -1)
|
atomic.AddInt32(&d.conns, -1)
|
||||||
}()
|
}()
|
||||||
if resp.StatusCode != http.StatusPartialContent {
|
if resp.StatusCode != http.StatusPartialContent {
|
||||||
return statusError(fmt.Sprintf("segment %d: expected 206, got %s", s.index, resp.Status), resp.StatusCode)
|
err := statusError(fmt.Sprintf("segment %d: expected 206, got %s", s.index, resp.Status), resp.StatusCode)
|
||||||
|
// A 200 means this mirror ignores Range entirely; retrying it only burns the
|
||||||
|
// budget, so make it fatal and let fetchSeg fail over to the next mirror.
|
||||||
|
if resp.StatusCode == http.StatusOK {
|
||||||
|
err = fatal{err}
|
||||||
|
}
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
// A mirror must serve the same file as the primary, or its bytes written at
|
// A mirror must serve the same file as the primary, or its bytes written at
|
||||||
// this offset would corrupt the output. Reject a 206 whose total length
|
// this offset would corrupt the output. Require a Content-Range that confirms
|
||||||
// disagrees with the probe (validate the total length); fatal so withRetries
|
// both the offset we asked for and the total length the probe saw; a missing,
|
||||||
// stops and fetchSeg falls over to the next mirror instead of writing it.
|
// unparseable, or divergent range is fatal so withRetries stops and fetchSeg
|
||||||
if _, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok && t != total {
|
// falls over to the next mirror instead of writing unverified bytes.
|
||||||
return fatal{fmt.Errorf("segment %d: %s reports length %d, want %d", s.index, uri, t, total)}
|
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)
|
body, stop := d.idleGuard(resp.Body, cancel)
|
||||||
defer stop()
|
defer stop()
|
||||||
return d.pump(ctx, f, p, s, body)
|
return d.pump(ctx, f, p, s, body)
|
||||||
}
|
}
|
||||||
|
|
||||||
// errIdleTimeout marks a read that stalled past the idle window, so callers can
|
// errIdleTimeout marks a transfer that stalled past the idle window, so callers
|
||||||
// report a distinct "timed out" instead of the bare "context canceled" that the
|
// can report a distinct "timed out" instead of the bare "context canceled" that
|
||||||
// cancelled request would otherwise surface (item [39]).
|
// the cancelled request would otherwise surface (item [39]).
|
||||||
var errIdleTimeout = errors.New("idle timeout: no data received")
|
var errIdleTimeout = errors.New("idle timeout: no data received")
|
||||||
|
|
||||||
// idleReader resets a watchdog timer on every read; if a read stalls longer
|
// idleReader records the time of the last byte received; a background watcher
|
||||||
// than the idle window the timer fires and cancels the request, so the blocked
|
// (idleGuard) cancels the request once a full idle window passes with no
|
||||||
// Read returns an error instead of hanging forever on a wedged connection.
|
// 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 {
|
type idleReader struct {
|
||||||
r io.Reader
|
r io.Reader
|
||||||
timer *time.Timer
|
|
||||||
idle time.Duration
|
idle time.Duration
|
||||||
fired atomic.Bool // set when the watchdog cancelled the request
|
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) {
|
func (ir *idleReader) Read(p []byte) (int, error) {
|
||||||
ir.timer.Reset(ir.idle)
|
|
||||||
n, err := ir.r.Read(p)
|
n, err := ir.r.Read(p)
|
||||||
ir.timer.Stop() // only the Read itself counts as idle, not write/throttle gaps
|
if n > 0 {
|
||||||
if err != nil && ir.fired.Load() {
|
ir.last.Store(time.Now().UnixNano())
|
||||||
// The cancellation came from our watchdog, not the caller's ctx, so
|
}
|
||||||
|
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.
|
// translate the read error into the distinct idle-timeout sentinel.
|
||||||
return n, fmt.Errorf("%w (after %s)", errIdleTimeout, ir.idle)
|
return n, fmt.Errorf("%w (after %s)", errIdleTimeout, ir.idle)
|
||||||
}
|
}
|
||||||
return n, err
|
return n, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// idleGuard wraps body with an idle timeout of d.cfg.Timeout, cancelling the
|
// watch cancels the request once idle passes with no byte received. It polls a
|
||||||
// request if no data arrives in that window. It returns the reader to use and a
|
// few times per window so a stall is caught within ~idle of the last byte,
|
||||||
// stop func to call when the transfer is done.
|
// 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()) {
|
func (d *Download) idleGuard(body io.Reader, cancel context.CancelFunc) (io.Reader, func()) {
|
||||||
if d.cfg.Timeout <= 0 {
|
if d.cfg.Timeout <= 0 {
|
||||||
return body, func() {}
|
return body, func() {}
|
||||||
}
|
}
|
||||||
ir := &idleReader{r: body, idle: d.cfg.Timeout}
|
ir := &idleReader{r: body, idle: d.cfg.Timeout}
|
||||||
ir.timer = time.AfterFunc(d.cfg.Timeout, func() {
|
ir.last.Store(time.Now().UnixNano())
|
||||||
ir.fired.Store(true)
|
stop := make(chan struct{})
|
||||||
cancel()
|
go ir.watch(stop, cancel)
|
||||||
})
|
return ir, func() { close(stop) }
|
||||||
return ir, func() { ir.timer.Stop() }
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// speedStartupGrace is how long a download may stay slow before the
|
// speedStartupGrace is how long a download may stay slow before the
|
||||||
@@ -1009,6 +1082,19 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
|
|||||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
|
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
|
||||||
return statusError(fmt.Sprintf("%s: %s", uri, resp.Status), resp.StatusCode)
|
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
|
// 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.
|
// over from the top rather than appending past the existing bytes.
|
||||||
if resp.StatusCode == http.StatusOK {
|
if resp.StatusCode == http.StatusOK {
|
||||||
|
|||||||
25
main.go
25
main.go
@@ -100,9 +100,15 @@ func run(args []string) int {
|
|||||||
eng := download.NewEngine(opts.Int("max-concurrent-downloads"))
|
eng := download.NewEngine(opts.Int("max-concurrent-downloads"))
|
||||||
|
|
||||||
uiCtx, uiCancel := context.WithCancel(context.Background())
|
uiCtx, uiCancel := context.WithCancel(context.Background())
|
||||||
|
uiDone := make(chan struct{})
|
||||||
if !opts.Bool("quiet") && !opts.Bool("dry-run") {
|
if !opts.Bool("quiet") && !opts.Bool("dry-run") {
|
||||||
rep := progress.New(eng.Snapshot, opts.Bool("human-readable"))
|
rep := progress.New(eng.Snapshot, opts.Bool("human-readable"))
|
||||||
go rep.Run(uiCtx)
|
go func() {
|
||||||
|
rep.Run(uiCtx)
|
||||||
|
close(uiDone)
|
||||||
|
}()
|
||||||
|
} else {
|
||||||
|
close(uiDone)
|
||||||
}
|
}
|
||||||
|
|
||||||
jobsRef.set(jobs)
|
jobsRef.set(jobs)
|
||||||
@@ -129,7 +135,7 @@ func run(args []string) int {
|
|||||||
}
|
}
|
||||||
|
|
||||||
uiCancel()
|
uiCancel()
|
||||||
time.Sleep(20 * time.Millisecond) // let the renderer clear its line
|
<-uiDone // wait for the renderer's final frame before printing the OK/FAIL lines
|
||||||
|
|
||||||
saveCookies(jobs)
|
saveCookies(jobs)
|
||||||
|
|
||||||
@@ -831,6 +837,11 @@ func makeLimiter(bps int64) *rate.Limiter {
|
|||||||
return rate.NewLimiter(rate.Limit(bps), download.LimiterBurst(bps))
|
return rate.NewLimiter(rate.Limit(bps), download.LimiterBurst(bps))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// maxSelectIndex bounds how far a --select-file range is expanded: a torrent has
|
||||||
|
// at most a few thousand files, so a degenerate spec like "1-9999999999" must not
|
||||||
|
// spin or balloon the map. Indexes past the real file count are ignored anyway.
|
||||||
|
const maxSelectIndex = 1 << 20
|
||||||
|
|
||||||
// parseSelect parses "1,3-5" into a set of 1-based file indexes.
|
// parseSelect parses "1,3-5" into a set of 1-based file indexes.
|
||||||
func parseSelect(s string) map[int]bool {
|
func parseSelect(s string) map[int]bool {
|
||||||
s = strings.TrimSpace(s)
|
s = strings.TrimSpace(s)
|
||||||
@@ -848,10 +859,14 @@ func parseSelect(s string) map[int]bool {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
b, _ := strconv.Atoi(strings.TrimSpace(hi))
|
b, _ := strconv.Atoi(strings.TrimSpace(hi))
|
||||||
for i := a; i <= b; i++ {
|
if a < 1 {
|
||||||
if i > 0 {
|
a = 1
|
||||||
out[i] = true
|
|
||||||
}
|
}
|
||||||
|
if b > maxSelectIndex {
|
||||||
|
b = maxSelectIndex
|
||||||
|
}
|
||||||
|
for i := a; i <= b; i++ {
|
||||||
|
out[i] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
|
|||||||
@@ -153,8 +153,7 @@ func (r *Reporter) lineOne(s download.Stat) string {
|
|||||||
fmt.Fprintf(&b, " UL:%s", speed(ul, r.human))
|
fmt.Fprintf(&b, " UL:%s", speed(ul, r.human))
|
||||||
}
|
}
|
||||||
if !seeding && s.Total > 0 && dl > 0 {
|
if !seeding && s.Total > 0 && dl > 0 {
|
||||||
eta := time.Duration(float64(s.Total-s.Completed)/float64(dl)) * time.Second
|
fmt.Fprintf(&b, " ETA:%s", secfmt(etaDuration(s.Total-s.Completed, dl)))
|
||||||
fmt.Fprintf(&b, " ETA:%s", secfmt(eta))
|
|
||||||
}
|
}
|
||||||
b.WriteByte(']')
|
b.WriteByte(']')
|
||||||
return b.String()
|
return b.String()
|
||||||
@@ -215,16 +214,23 @@ func (r *Reporter) summary(stats []download.Stat) string {
|
|||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
var totDL, totUL int64
|
var totDL, totUL int64
|
||||||
stalled := 0
|
stalled, waiting := 0, 0
|
||||||
for _, s := range stats {
|
for _, s := range stats {
|
||||||
dl, ul := r.rates(s)
|
dl, ul := r.rates(s)
|
||||||
totDL += dl
|
totDL += dl
|
||||||
totUL += ul
|
totUL += ul
|
||||||
|
if s.Status == download.Waiting {
|
||||||
|
waiting++
|
||||||
|
continue
|
||||||
|
}
|
||||||
if r.isStalled(s, dl) {
|
if r.isStalled(s, dl) {
|
||||||
stalled++
|
stalled++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
line := fmt.Sprintf("%d active DL:%s UL:%s", len(stats), speed(totDL, r.human), speed(totUL, r.human))
|
line := fmt.Sprintf("%d active DL:%s UL:%s", len(stats)-waiting, speed(totDL, r.human), speed(totUL, r.human))
|
||||||
|
if waiting > 0 {
|
||||||
|
line += fmt.Sprintf(" (%d waiting)", waiting)
|
||||||
|
}
|
||||||
if stalled > 0 {
|
if stalled > 0 {
|
||||||
line += fmt.Sprintf(" (%d stalled)", stalled)
|
line += fmt.Sprintf(" (%d stalled)", stalled)
|
||||||
}
|
}
|
||||||
@@ -245,6 +251,8 @@ func (r *Reporter) tableRow(s download.Stat) string {
|
|||||||
// upload rate, or a "stalled" flag).
|
// upload rate, or a "stalled" flag).
|
||||||
func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string) {
|
func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string) {
|
||||||
switch s.Status {
|
switch s.Status {
|
||||||
|
case download.Waiting:
|
||||||
|
return "queued", ""
|
||||||
case download.Seeding:
|
case download.Seeding:
|
||||||
return "seeding", "UL:" + speed(ul, r.human)
|
return "seeding", "UL:" + speed(ul, r.human)
|
||||||
case download.Complete:
|
case download.Complete:
|
||||||
@@ -255,8 +263,7 @@ func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string
|
|||||||
metric = "DL:" + speed(dl, r.human)
|
metric = "DL:" + speed(dl, r.human)
|
||||||
switch {
|
switch {
|
||||||
case dl > 0 && s.Total > 0:
|
case dl > 0 && s.Total > 0:
|
||||||
eta := time.Duration(float64(s.Total-s.Completed)/float64(dl)) * time.Second
|
tail = secfmt(etaDuration(s.Total-s.Completed, dl))
|
||||||
tail = secfmt(eta)
|
|
||||||
case r.isStalled(s, dl):
|
case r.isStalled(s, dl):
|
||||||
tail = "stalled"
|
tail = "stalled"
|
||||||
}
|
}
|
||||||
@@ -264,6 +271,20 @@ func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// maxETASeconds caps an ETA before it becomes a Duration: a near-stalled download
|
||||||
|
// yields an astronomically large seconds value, and time.Duration(secs) *
|
||||||
|
// time.Second would overflow int64 and wrap to a bogus tiny ETA. secfmt renders
|
||||||
|
// anything past 99h as "--", so clamping to 100h loses nothing.
|
||||||
|
const maxETASeconds = 100 * 3600
|
||||||
|
|
||||||
|
func etaDuration(remaining, dl int64) time.Duration {
|
||||||
|
secs := float64(remaining) / float64(dl)
|
||||||
|
if secs > maxETASeconds {
|
||||||
|
secs = maxETASeconds
|
||||||
|
}
|
||||||
|
return time.Duration(secs) * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
// isStalled reports an active download that has gone a while with no new bytes,
|
// isStalled reports an active download that has gone a while with no new bytes,
|
||||||
// so the table can flag it rather than leave the eye to read DL:0B. A download
|
// so the table can flag it rather than leave the eye to read DL:0B. A download
|
||||||
// too young to have a rate yet is not stalled.
|
// too young to have a rate yet is not stalled.
|
||||||
|
|||||||
Reference in New Issue
Block a user