diff --git a/download/engine.go b/download/engine.go index 39c085f..75e3094 100644 --- a/download/engine.go +++ b/download/engine.go @@ -91,6 +91,10 @@ func (e *Engine) track(d Download) { e.mu.Unlock() } +// untrack removes d from the active set. Every Download implementation is a +// pointer type, so the == compares identities — never the fields of a value, +// which could panic on a non-comparable one. (Keying on Stat().ID is not an +// option: a torrent's ID changes from source to infohash once metadata loads.) func (e *Engine) untrack(d Download) { e.mu.Lock() defer e.mu.Unlock() diff --git a/httpdl/httpdl.go b/httpdl/httpdl.go index 10dc6a3..06a5234 100644 --- a/httpdl/httpdl.go +++ b/httpdl/httpdl.go @@ -536,29 +536,43 @@ func (d *Download) withRetries(ctx context.Context, what string, attempt func() 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 lastErr error - for _, u := range d.uris { - if ctx.Err() != nil { - return probeResult{}, ctx.Err() - } - var pr probeResult - err := d.withRetries(ctx, u, func() error { + 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, u) + pr, perr = d.probe(ctx, uri) return perr }) - if err == nil { - return pr, nil - } - lastErr = err + }) + if err != nil { + return probeResult{}, err } - return probeResult{}, lastErr + return pr, nil } // probe issues a one-byte ranged GET to learn the total size, whether the @@ -715,24 +729,14 @@ func (d *Download) fetchSeg(ctx context.Context, f *os.File, p *pool, s *seg, to // 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(). - var lastErr error - for i := range d.uris { - if ctx.Err() != nil { - return ctx.Err() - } - uri := d.mirror(s.index + i) - err := d.withRetries(ctx, fmt.Sprintf("segment %d", s.index), func() error { + 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, p, s, uri, total) }) - if err == nil { - return nil - } - lastErr = err - } - return lastErr + }) } func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, uri string, total int64) error { @@ -816,10 +820,9 @@ 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: d.completed is not -// strictly monotonic — singleOnce rewrites it with an absolute StoreInt64 and a -// resumed Range answered by a 200 resets the offset to 0, so the counter can -// jump backward on a retry. Treating that backward jump as "below limit" would +// 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. @@ -964,23 +967,13 @@ func (d *Download) single(ctx context.Context, out string) error { 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). - var lastErr error - for i := range d.uris { - if ctx.Err() != nil { - return ctx.Err() - } - uri := d.mirror(i) - err := d.withRetries(ctx, uri, func() error { + 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) }) - if err == nil { - return nil - } - lastErr = err - } - return lastErr + }) } // singleOnce performs one single-stream transfer attempt. When resumeFromDisk is @@ -1058,7 +1051,7 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi } off += int64(rd) got += int64(rd) - atomic.StoreInt64(&d.completed, off) + atomic.AddInt64(&d.completed, int64(rd)) // add deltas, as pump does d.throttle(ctx, rd) } if err == io.EOF { diff --git a/main.go b/main.go index 5eee736..d067fd7 100644 --- a/main.go +++ b/main.go @@ -190,9 +190,9 @@ type failedDownload struct { err error } -func (f failedDownload) Name() string { return f.name } -func (f failedDownload) Run(context.Context) error { return f.err } -func (f failedDownload) Stat() download.Stat { +func (f *failedDownload) Name() string { return f.name } +func (f *failedDownload) Run(context.Context) error { return f.err } +func (f *failedDownload) Stat() download.Stat { return download.Stat{Name: f.name, Status: download.Errored, Total: -1} } @@ -281,7 +281,7 @@ func build(opts *cli.Options, targets []target) ([]job, *torrent.Client, error) u := t.uris[0] // primary: classifies the target and names the session source src := strings.Join(t.uris, "\t") if first, isDup := dups[u]; isDup { - out = append(out, job{source: src, dl: failedDownload{ + out = append(out, job{source: src, dl: &failedDownload{ name: u, err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first), }}) @@ -397,7 +397,7 @@ func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job { // from under the other, stranding it at 0%. build() dedupes pass-1 sources // but cannot reach a not-yet-fetched HTTP .torrent, so dedupe here too. if first, dup := markInfoHash(seen, path); dup { - out = append(out, job{source: path, dl: failedDownload{ + out = append(out, job{source: path, dl: &failedDownload{ name: path, err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first), }})