Compare commits

...

10 Commits

Author SHA1 Message Date
81c44d9ec7 httpdl/cli: drop --auto-split; classify errors by type, not message text
Two Pike cleanups.

--auto-split was a got-only third connection-count policy (beside -x and -s)
that invented rather than matched aria2. Removing it also retires the now-dead
autoConns/maxAutoConns and the per-host transport cap that existed only to scale
for it; min(split, M*-x) is the sole policy again.

main's exit-code mapping fell back to substring-matching third-party error text
(strings.Contains "connection refused"/"timeout"/...), which rots when a
dependency rewords a message. The typed checks (errors.Is on the syscall errno,
*net.DNSError, net.Error.Timeout, context.DeadlineExceeded) already cover the
real stdlib errors -- verified end to end: refused -> 6, bad host -> 19. The two
cases that genuinely needed the text match are our own errors, now typed
sentinels: httpdl.ErrTimeout (idle timeout) and the bt metadata timeout wrapping
context.DeadlineExceeded.
2026-06-20 23:55:39 +09:00
270812de3e httpdl: drop work-stealing segment pool for one worker per segment
The pool let an idle worker steal the back half of a slower in-flight
segment, behind a mutex that also serialized every advance. At the default
-x 1 it never fired, and it carried the trickiest invariant in the tree
(minSplit >= readBuf keeps an owner's in-flight write out of the stolen tail).

makeSegments already caps the segment count at the connection count, so the
segments map one-to-one onto workers: give each its own goroutine, advance
written with a plain atomic add, and snapshot the fixed slice with no lock.
pool/newPool/acquire/steal/advance and the seg.owned field all go away.

Net -156 lines. Behaviour at the user's flags is unchanged, except a slow
mirror's tail segment is no longer rebalanced onto idle connections.
2026-06-20 23:43:15 +09:00
65104ade92 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
2026-06-20 23:28:26 +09:00
db73b51e0d gitignore: ignore downloaded media so an in-tree run can't commit content
running got in the repo root drops downloaded files next to the source;
torrents/control files were already ignored, add common media (mkv/mp4/
avi/mov/webm/iso/mp3/flac) so anime/ISO downloads don't show up as
untracked and can't be committed by accident.
2026-06-20 21:54:29 +09:00
27a5cece41 cli: show -x/-s/-k in basic help, matching aria2
max-connection-per-server, split and min-split-size were tagged HTTP, so
they only appeared under --help=all. aria2 tags all three TAG_BASIC; tag
them Basic too so they show in the default --help, where users expect the
core connection knobs.
2026-06-20 21:53:33 +09:00
d9de63f451 httpdl: unify completed counter and dedupe mirror loop; engine: harden untrack
From a Rob Pike pass over the tree:
- singleOnce adds deltas to d.completed like pump does, instead of
  storing absolute offsets, so the counter has one write discipline.
- one overMirrors helper replaces the three copy-pasted mirror-fallover
  loops in probeRetry, fetchSeg and single.
- failedDownload becomes a pointer, the only value-type Download impl, so
  Engine.untrack's == is always pointer identity and can't compare the
  fields of a value (keying on Stat().ID is out — a torrent's ID changes
  from source to infohash once metadata loads).
2026-06-20 21:12:07 +09:00
11734af7fb bt: filter anacrolix logs to errors so warnings don't corrupt the readout
webseed/peer/tracker warnings (the 403/429 chatter) go through slog, the
rest through the legacy analog logger; both default to Warning, which
interleaves with and corrupts the live progress block. Filter both to
Error and above so only genuine errors reach stderr. quietSlogger is a
small testable helper; anacrolix/log becomes a direct dependency.
2026-06-20 21:11:57 +09:00
16cbb97748 progress: show multiple downloads as a bar table
lineMany crammed every download onto one line as name+percent, dropping
the speed column that answers "which one is stalled" and producing ugly
double brackets for names like "[Gecko]". render N>1 as a redrawn block:
one row each (name, bar, percent, DL speed, ETA/stalled), a header with
the active count and totals, falling back to a single summary line when
the table would overflow the window. one download keeps its dashboard.
2026-06-20 19:45:49 +09:00
ee66235c36 main: dedupe followed .torrent files by infohash
a .torrent fetched over http is added to the shared client in a second
pass (followUps); two that name the same torrent would share one torrent
on the refcount-less client, and the first to finish would Drop it out
from under the other, stranding it at 0%. dedupe the followed files by
infohash, the guard build() already applies to pass-1 sources.
2026-06-20 19:28:32 +09:00
3d9ea7ce8b httpdl: add segment work-stealing and --auto-split
an idle worker now steals the back half of the busiest in-flight segment
instead of exiting, so one slow segment no longer drains alone over a single
connection. control file persists stolen tails; resume re-tiles.

add --auto-split (got-only, opt-in): derive the connection count from file
size (one per min-split-size, up to 16), overriding -x/-s. default stays
aria2-faithful at 1 connection for a single server.
2026-06-19 14:01:59 +09:00
17 changed files with 929 additions and 365 deletions

10
.gitignore vendored
View File

@@ -7,3 +7,13 @@ ref/
bin/
.torrent.db*
*.part
# media downloaded by running got in the repo tree (never commit content)
*.mkv
*.mp4
*.avi
*.mov
*.webm
*.iso
*.mp3
*.flac

150
README.md
View File

@@ -1,13 +1,8 @@
# got
A small, Rob-Pike-style command-line download tool in Go: one program that
downloads HTTP(S) URLs over several connections and BitTorrent
magnets/torrents, with a focused set of options.
The design leans on Go's runtime instead of a hand-rolled reactor. There is no
single-threaded event-poll loop: one goroutine per download blocks on real I/O,
a buffered channel of slots bounds concurrency, and `context` carries shutdown.
What remains is the data — and, per Pike, the data is the design.
A command-line download tool in Go. It fetches HTTP(S) files over multiple
connections and downloads BitTorrent magnets and `.torrent` files — one binary,
familiar flags.
## Build
@@ -15,148 +10,63 @@ What remains is the data — and, per Pike, the data is the design.
go build -o got .
```
Requires Go 1.25+. The BitTorrent engine is
[`anacrolix/torrent`](https://github.com/anacrolix/torrent); everything else is
the standard library plus `golang.org/x/{term,time}`.
Needs Go 1.25+.
## Use
## Usage
```sh
# segmented HTTP download, 16 connections, into ./downloads
got -x16 -s16 -d downloads https://example.com/big.iso
# HTTP download with 16 connections
got -x16 -s16 https://example.com/big.iso
# resume an interrupted download
got -c -x16 https://example.com/big.iso
got -c https://example.com/big.iso
# the same file from several mirrors at once (one file, connections spread
# across servers, a dead mirror falls over); -Z downloads them separately instead
# one file from several mirrors (connections spread across servers,
# a dead mirror falls over); -Z downloads them as separate files instead
got -x4 https://a.example/big.iso https://b.example/big.iso
# a torrent or magnet; seed for 30 minutes after finishing
got --seed-time=30 ubuntu.torrent
got 'magnet:?xt=urn:btih:...'
# list the files in a torrent without downloading
got -S -T some.torrent
# several downloads at once, two running in parallel
# many downloads, two at a time
got -j2 -i urls.txt
```
Run `got --help` (or `--help=http`, `--help=bittorrent`, ...) for the full
option list.
Run `got --help` (or `--help=all`) for every option.
### Options
A focused subset of options, with familiar short flags and sensible defaults.
## Common options
| flag | meaning | default |
|------|---------|---------|
| `-d, --dir` | output directory | `.` |
| `-o, --out` | output filename (single download) | from URL |
| `-i, --input-file` | read downloads line by line (TAB-separated URLs = mirrors; `-` = stdin) | |
| `-o, --out` | output filename | from URL |
| `-c, --continue` | resume a partial download | false |
| `-x, --max-connection-per-server` | connections to one server (116) | 1 |
| `-s, --split` | split a download into N connections | 5 |
| `-Z, --force-sequential` | download each command-line URL separately, not as mirrors | false |
| `-k, --min-split-size` | do not split below this size | 20M |
| `-j, --max-concurrent-downloads` | parallel downloads | 5 |
| `-j, --max-concurrent-downloads` | downloads at once | 5 |
| `--max-overall-download-limit` | global speed cap | 0 (off) |
| `--max-download-limit` | per-download speed cap | 0 (off) |
| `--retry-wait` | seconds to wait between retries | 0 |
| `--load-cookies` / `--save-cookies` | read/write a Netscape `cookies.txt` | |
| `--conditional-get` | skip the download if the local file is up to date | false |
| `--remote-time` | set the file's mtime from the server | false |
| `--checksum` | verify the finished file: `TYPE=DIGEST` (sha-256, sha-1, md5, …) | |
| `--dry-run` | check the file is available but do not download it | false |
| `--ca-certificate` | verify HTTPS against the CA certificates in FILE (PEM) | |
| `--stop` | stop the program after N seconds | 0 (off) |
| `--disable-ipv6` | force IPv4-only connections | false |
| `-T, --torrent-file` | a `.torrent` file | |
| `--seed-time` | minutes to seed after finishing | seed by ratio |
| `--checksum` | verify the finished file: `TYPE=DIGEST` (sha-256, sha-1, …) | |
| `--seed-time` | minutes to seed after a torrent finishes | by ratio |
| `--seed-ratio` | stop seeding at this ratio | 1.0 |
| `--listen-port` | port (range) for incoming peers | 6881-6999 |
| `--enable-dht` | use the BitTorrent DHT | true |
| `--select-file` | fetch only these file indexes (`1,3-5`) | all |
| `-S, --show-files` | list torrent files and exit | |
| `--bt-stop-timeout` | give up if no download progress for N s | 0 (off) |
| `--bt-metadata-timeout` | give up if a magnet can't fetch metadata in N s | 60 |
| `--save-session` | on exit, write unfinished downloads to FILE | |
| `-q, --quiet` | no progress readout | false |
| `-q, --quiet` | no progress output | false |
### Resuming across runs
## Resume
There is no daemon — resume is a plain file. Point
`--save-session` and `-i` at the same file and add `-c`: unfinished downloads
are loaded at start and the still-unfinished ones are written back on exit.
HTTP downloads resume from a small `<file>.got` sidecar — just re-run with `-c`.
Torrents resume from the data already on disk. To resume across runs, point
`--save-session` and `-i` at the same file:
```sh
got -c --save-session=got.session -i got.session <new urls/magnets...>
got -c --save-session=got.session -i got.session <urls/magnets...>
```
On the first run the session file need not exist. HTTP downloads resume from
their `.got` control file; torrents resume from the data already on disk
(re-checked by the engine). A magnet with no reachable peers no longer hangs:
`--bt-metadata-timeout` bounds the metadata fetch and defaults to 60s (set it to
0 to wait forever).
## Notes
## Design
```
main classify URIs, wire options, run, report exit code
└─ download the Download interface + the scheduler (-j semaphore, ctx)
├─ httpdl []Segment + worker pool + WriteAt + JSON resume sidecar
└─ bt anacrolix/torrent behind the Download interface
├─ cli one flat option table -> hand parser -> layered config
└─ progress one ticker goroutine, pull-snapshot, \r line redraw
```
The contract is one small interface:
```go
type Download interface {
Name() string
Run(ctx context.Context) error // blocks in its own goroutine
Stat() Stat // a value snapshot, lock-free to read
}
```
- **HTTP** splits the file into byte-range `Segment`s handed to a pool of
workers. Each worker streams its range straight to one shared file with
`os.File.WriteAt` — safe for concurrent non-overlapping writes, so there is no
shared seek offset and no mutex. Resume is a small JSON sidecar
(`<file>.got`) holding per-segment progress plus validators (length +
ETag/Last-Modified) so a stale file is never trusted.
- **BitTorrent** is `anacrolix/torrent` configured from the CLI options and
driven to completion, then seeded for `--seed-time` minutes or up to
`--seed-ratio`.
- **Options** are one flat `[]Opt` table — the single source of truth for
parsing, validation, defaults and `--help`. Layers apply in order:
built-in defaults, config file, proxy environment, command line.
- **Progress** is a single goroutine that ticks once a second, pulls a snapshot
of the running downloads, derives speeds from the change since the last tick,
and redraws one line with `\r` + erase-to-end-of-line.
## Status
Implemented: segmented HTTP(S) with resume — including a foreign/browser-started
partial file (`-c`) — multi-mirror downloads (several URLs, or TAB-grouped `-i`
lines, fetch one file with connections spread across servers and fallover; `-Z`
to download them separately), retries, rate limits, auto-renaming,
`Content-Disposition` naming, conditional GET, cookies
(`--load-cookies`/`--save-cookies`), HTTP basic auth, `--remote-time`, whole-file
checksum verification (`--checksum`), and a `--dry-run` availability check;
BitTorrent download + seeding (magnet and `.torrent`,
DHT, trackers, file selection, show-files, metadata-fetch timeout), seeding that
stops on whichever of `--seed-time`/`--seed-ratio` comes first, and a listen port
chosen from the whole range so a busy port no longer disables BitTorrent;
a command-line interface with documented exit codes, config file, live progress,
and session save/reload for resume across runs.
Deferred (not implemented yet): FTP/SFTP, Metalink, and the
JSON-RPC server. The architecture leaves room for
each — a new protocol is just another `Download`. One known limitation: the
BitTorrent engine (`anacrolix/torrent`) applies download/upload rate limits
client-wide, so `--max-upload-limit`/`--max-download-limit` act per-run rather
than per-torrent when several torrents run at once (`--max-overall-*` are exact);
DHT/PEX likewise cannot yet be disabled per private torrent.
- BitTorrent uses [`anacrolix/torrent`](https://github.com/anacrolix/torrent);
HTTP uses the standard library.
- Not implemented: FTP/SFTP, Metalink, and the JSON-RPC server.
- When several torrents run at once, `--max-download-limit` /
`--max-upload-limit` apply per run rather than per torrent (`--max-overall-*`
are exact).

View File

@@ -9,12 +9,16 @@ package bt
import (
"context"
"fmt"
"io"
"log/slog"
"os"
"path/filepath"
"strconv"
"strings"
"sync/atomic"
"time"
alog "github.com/anacrolix/log"
missinggo "github.com/anacrolix/missinggo/v2"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/metainfo"
@@ -64,15 +68,30 @@ func ParsePorts(spec string) []int {
if err != nil {
continue
}
// Clamp to the valid port space before iterating: a parseable but oversized
// spec like "1-9999999999" would otherwise spin for ~10^10 iterations and
// 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++ {
if i >= 1 && i <= 65535 {
ports = append(ports, i)
}
}
}
return ports
}
// quietSlogger filters anacrolix's structured logging down to Error and above,
// so the routine webseed/peer/tracker warning chatter (the 403/429 noise) no
// longer interleaves with — and corrupts — the live progress display. Genuine
// errors still reach w.
func quietSlogger(w io.Writer) *slog.Logger {
return slog.New(slog.NewTextHandler(w, &slog.HandlerOptions{Level: slog.LevelError}))
}
// NewClient builds the shared anacrolix client from the run-wide config. When a
// listen-port spec is given it tries each candidate port in turn, advancing past
// any that is already in use, and falls back to an OS-chosen port if none bind,
@@ -91,6 +110,11 @@ func ParsePorts(spec string) []int {
// prevention must come from a run-wide setting (e.g. --enable-dht=false).
func NewClient(cfg ClientConfig) (*torrent.Client, error) {
c := torrent.NewDefaultClientConfig()
// Keep the library's own logging out of the readout. Webseed/peer warnings go
// through slog, the rest through the legacy analog logger, so filter both to
// Error and above (the analog default is Warning, which is what leaks today).
c.Slogger = quietSlogger(os.Stderr)
c.Logger = alog.Default.FilterLevel(alog.Error)
c.DataDir = cfg.Dir
if c.DataDir == "" {
c.DataDir = "."
@@ -289,14 +313,17 @@ func (d *Download) Run(ctx context.Context) (err error) {
return nil
}
if err := d.choose(t); err != nil {
return err
}
// --check-integrity: re-hash the existing on-disk data BEFORE arming the
// 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 err := t.VerifyDataContext(ctx); err != nil {
return err
}
}
if err := d.choose(t); err != nil {
return err
}
if err := d.wait(ctx, t); err != nil {
return err
@@ -365,7 +392,9 @@ func (d *Download) awaitInfo(ctx context.Context, t *torrent.Torrent) error {
case <-ctx.Done():
return ctx.Err()
case <-deadline:
return fmt.Errorf("timed out fetching metadata after %s", d.opts.MetaTimeout)
// Wrap DeadlineExceeded so the exit-code mapping classifies it as a timeout
// (2) via errors.Is, rather than matching on the message text.
return fmt.Errorf("timed out fetching metadata after %s: %w", d.opts.MetaTimeout, context.DeadlineExceeded)
}
}

View File

@@ -1,10 +1,38 @@
package bt
import (
"bytes"
"context"
"log/slog"
"reflect"
"strings"
"testing"
)
// TestQuietSloggerDropsWarnings: the library's warning chatter (the webseed
// 403/429 noise) is filtered out so it can't corrupt the live display, while a
// genuine error still gets through.
func TestQuietSloggerDropsWarnings(t *testing.T) {
var buf bytes.Buffer
lg := quietSlogger(&buf)
lg.Warn("webseed request error", "err", "429 Too Many Requests")
if buf.Len() != 0 {
t.Errorf("warning leaked into output: %q", buf.String())
}
// anacrolix's webseed code logs via Log(ctx, level, ...); confirm that path is
// filtered too, not just the Warn helper.
lg.Log(context.Background(), slog.LevelWarn, "another warning")
if buf.Len() != 0 {
t.Errorf("Log-at-Warn leaked: %q", buf.String())
}
lg.Error("real failure")
if !strings.Contains(buf.String(), "real failure") {
t.Errorf("error was suppressed, want it kept: %q", buf.String())
}
}
func TestParsePorts(t *testing.T) {
tests := []struct {
spec string

View File

@@ -97,7 +97,12 @@ func flagLabel(o *Opt) string {
case Int:
b.WriteString("=N")
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")
}
case Enum:
b.WriteString("=" + strings.Join(o.Enum, "|"))
default:

View File

@@ -78,9 +78,9 @@ var options = []Opt{
{Long: "human-readable", Kind: Bool, Default: "true", Help: "show sizes as Ki/Mi/Gi", Tag: Basic},
// --- http ---
{Long: "max-connection-per-server", Short: 'x', Kind: Int, Default: "1", Min: 1, Max: 16, Help: "max connections to one server (1-16)", Tag: HTTP},
{Long: "split", Short: 's', Kind: Int, Default: "5", Min: 1, Help: "split a download into N connections; actual connections are min(max-connection-per-server, split), and -x defaults to 1", Tag: HTTP},
{Long: "min-split-size", Short: 'k', Kind: Size, Default: "20M", Min: 1 << 20, Max: 1 << 30, Help: "do not split a piece smaller than SIZE (1M-1024M)", Tag: HTTP},
{Long: "max-connection-per-server", Short: 'x', Kind: Int, Default: "1", Min: 1, Max: 16, Help: "max connections to one server (1-16)", Tag: Basic},
{Long: "split", Short: 's', Kind: Int, Default: "5", Min: 1, Help: "split a download into N connections; actual connections are min(max-connection-per-server, split), and -x defaults to 1", Tag: Basic},
{Long: "min-split-size", Short: 'k', Kind: Size, Default: "20M", Min: 1 << 20, Max: 1 << 30, Help: "do not split a piece smaller than SIZE (1M-1024M)", Tag: Basic},
{Long: "force-sequential", Short: 'Z', Kind: Bool, Default: "false", Help: "download each command-line URI as its own file instead of mirroring them", Tag: HTTP},
{Long: "max-tries", Short: 'm', Kind: Int, Default: "5", Min: 0, Help: "max retries per segment (0 = unlimited)", Tag: HTTP},
{Long: "timeout", Short: 't', Kind: Int, Default: "60", Min: 1, Help: "connection timeout in seconds", Tag: HTTP},

View File

@@ -2,6 +2,7 @@ package cli
import (
"fmt"
"math"
"strconv"
"strings"
)
@@ -105,5 +106,10 @@ func parseSize(s string) (int64, error) {
if n < 0 {
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
}

View File

@@ -46,6 +46,12 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
go func(i int, d Download) {
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
// before this download ever started.
select {
@@ -56,9 +62,6 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
}
defer func() { <-slots }()
e.track(d)
defer e.untrack(d)
err := d.Run(ctx)
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
}(i, d)
@@ -74,7 +77,8 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
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 {
e.mu.Lock()
defer e.mu.Unlock()
@@ -91,6 +95,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()

2
go.mod
View File

@@ -3,6 +3,7 @@ module github.com/hanbok/got
go 1.25
require (
github.com/anacrolix/log v0.17.1-0.20251118025802-918f1157b7bb
github.com/anacrolix/missinggo/v2 v2.10.0
github.com/anacrolix/torrent v1.61.0
golang.org/x/sys v0.38.0
@@ -19,7 +20,6 @@ require (
github.com/anacrolix/envpprof v1.4.0 // indirect
github.com/anacrolix/generics v0.1.1-0.20251125230353-15d98d46693b // indirect
github.com/anacrolix/go-libutp v1.3.2 // indirect
github.com/anacrolix/log v0.17.1-0.20251118025802-918f1157b7bb // indirect
github.com/anacrolix/missinggo v1.3.0 // indirect
github.com/anacrolix/missinggo/perf v1.0.0 // indirect
github.com/anacrolix/mmsg v1.0.1 // indirect

View File

@@ -4,6 +4,7 @@ import (
"encoding/json"
"os"
"path/filepath"
"sort"
"sync/atomic"
)
@@ -27,11 +28,13 @@ type segState struct {
func controlPath(out string) string { return out + ".got" }
// snapshot builds a control record from the live segments.
// snapshot builds a control record from the live segments. The segment slice is
// fixed once the file is divided and each segment's frontier is owned by one
// worker, so reading written atomically gives a consistent record with no lock.
func snapshot(url string, total int64, etag, lastmod string, segs []seg) control {
c := control{URL: url, Total: total, ETag: etag, LastModified: lastmod, Segs: make([]segState, len(segs))}
for i := range segs {
c.Segs[i] = segState{segs[i].start, segs[i].end, atomic.LoadInt64(&segs[i].written)}
c.Segs[i] = segState{segs[i].start, segs[i].endOff(), atomic.LoadInt64(&segs[i].written)}
}
return c
}
@@ -73,7 +76,35 @@ func loadControl(out, url string, total int64, etag, lastmod string) *control {
if (lastmod != "" || c.LastModified != "") && c.LastModified != lastmod {
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
}
// 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)) }

View File

@@ -1,6 +1,6 @@
// 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, hand them to a pool of worker goroutines, and have each worker
// 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
@@ -119,19 +119,20 @@ type Config struct {
// connections.
type Download struct {
uris []string // mirror list; uris[0] is the primary (naming + resume key)
maxConns int // segment workers = min(split, len(uris)*max-connection-per-server)
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), then
// never mutated again, so Stat can read name from another goroutine without a
// lock once Run has set it. They are seeded in New so a Stat before Run still
// has a sensible best-effort name.
name string
out string
// 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
@@ -224,7 +225,21 @@ func New(uris []string, cfg Config) *Download {
TLSClientConfig: tlsCfg,
}
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)
}
}
@@ -286,29 +301,47 @@ func New(uris []string, cfg Config) *Download {
client.Jar = jar
}
return &Download{
d := &Download{
uris: uris,
maxConns: maxConns,
cfg: cfg,
client: client,
limit: limit,
jar: jar,
name: name,
out: out,
total: -1,
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).
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 {
return download.Stat{
Name: d.name,
ID: d.out, // resolved output path; stable per process (CONTRACT)
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),
@@ -392,15 +425,19 @@ func (d *Download) Run(ctx context.Context) (err error) {
// 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, d.out)
err = d.single(xferCtx, out)
} 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()
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 err
@@ -408,14 +445,14 @@ func (d *Download) Run(ctx context.Context) (err error) {
// --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(d.out, newSum, sumWant); err != 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(d.out, t, t)
os.Chtimes(out, t, t)
}
}
d.setStatus(download.Complete)
@@ -427,6 +464,7 @@ func (d *Download) Run(ctx context.Context) (err error) {
// 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 != "" {
@@ -438,8 +476,9 @@ func (d *Download) resolveName(pr probeResult) error {
if name == "" {
name = nameFromURL(d.primary())
}
d.name = name
d.out = filepath.Join(d.cfg.Dir, name)
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
@@ -450,20 +489,20 @@ func (d *Download) resolveName(pr probeResult) error {
// -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(d.out) {
if d.cfg.Continue && fileExists(out) {
return nil
}
if fileExists(d.out) && !d.cfg.AllowOverwrite {
if fileExists(out) && !d.cfg.AllowOverwrite {
if d.cfg.AutoRename {
u, err := uniqueName(d.out)
u, err := uniqueName(out)
if err != nil {
return err
}
d.out = u
d.name = filepath.Base(u)
d.setOut(u)
d.setName(filepath.Base(u))
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
}
@@ -528,30 +567,44 @@ 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 {
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 {
})
if err != nil {
return probeResult{}, err
}
return pr, nil
}
lastErr = err
}
return probeResult{}, lastErr
}
// 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
@@ -565,8 +618,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
// in-progress resume must not be short-circuited), using the local file's
// mtime for If-Modified-Since.
if d.cfg.ConditionalGet && !fileExists(controlPath(d.out)) {
if fi, err := os.Stat(d.out); err == nil {
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))
}
}
@@ -652,37 +705,23 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
saveDone := make(chan struct{})
go d.saveLoop(ctx, out, total, etag, lastmod, segs, saveDone)
jobs := make(chan *seg)
go func() {
defer close(jobs)
for i := range segs {
if segs[i].done() {
continue
}
select {
case jobs <- &segs[i]:
case <-ctx.Done(): // stop feeding once we're tearing down
return
}
}
}()
// 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 w := 0; w < d.maxConns; w++ {
for i := range segs {
wg.Add(1)
go func() {
go func(s *seg) {
defer wg.Done()
for s := range jobs {
if err := d.fetchSeg(ctx, f, s, total); err != nil {
errOnce.Do(func() { runErr = err; cancel() })
return
}
}
}()
}(&segs[i])
}
wg.Wait()
cancel()
@@ -706,30 +745,20 @@ func (d *Download) fetchSeg(ctx context.Context, f *os.File, s *seg, total int64
// 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, s, uri, total)
})
if err == nil {
return nil
}
lastErr = err
}
return lastErr
})
}
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))
req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.endOff()))
if err != nil {
return err
}
@@ -744,60 +773,95 @@ func (d *Download) fetchOnce(ctx context.Context, f *os.File, s *seg, uri string
atomic.AddInt32(&d.conns, -1)
}()
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
// this offset would corrupt the output. Reject a 206 whose total length
// disagrees with the probe (validate the total length); fatal so withRetries
// stops and fetchSeg falls over to the next mirror instead of writing it.
if _, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok && t != total {
return fatal{fmt.Errorf("segment %d: %s reports length %d, want %d", s.index, uri, t, total)}
// 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)
}
// errIdleTimeout marks a read that stalled past the idle window, so callers can
// report a distinct "timed out" instead of the bare "context canceled" that the
// cancelled request would otherwise surface (item [39]).
var errIdleTimeout = errors.New("idle timeout: no data received")
// 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 resets a watchdog timer on every read; if a read stalls longer
// than the idle window the timer fires and cancels the request, so the blocked
// Read returns an error instead of hanging forever on a wedged connection.
// 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
timer *time.Timer
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) {
ir.timer.Reset(ir.idle)
n, err := ir.r.Read(p)
ir.timer.Stop() // only the Read itself counts as idle, not write/throttle gaps
if err != nil && ir.fired.Load() {
// The cancellation came from our watchdog, not the caller's ctx, so
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)", errIdleTimeout, ir.idle)
return n, fmt.Errorf("%w (after %s)", ErrTimeout, ir.idle)
}
return n, err
}
// idleGuard wraps body with an idle timeout of d.cfg.Timeout, cancelling the
// request if no data arrives in that window. It returns the reader to use and a
// stop func to call when the transfer is done.
// 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.timer = time.AfterFunc(d.cfg.Timeout, func() {
ir.fired.Store(true)
cancel()
})
return ir, func() { ir.timer.Stop() }
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
@@ -807,10 +871,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.
@@ -901,7 +964,9 @@ func statusError(ctxMsg string, code int) error {
}
// pump copies the response body into the file at the segment's running offset,
// stopping at the segment end and respecting rate limits.
// 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 {
@@ -914,7 +979,7 @@ func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader)
if _, werr := f.WriteAt(buf[:rd], s.offset()); werr != nil {
return werr
}
s.advance(int64(rd))
s.addWritten(int64(rd))
atomic.AddInt64(&d.completed, int64(rd))
d.throttle(ctx, rd)
}
@@ -953,23 +1018,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
@@ -1005,6 +1060,19 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
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 {
@@ -1047,7 +1115,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 {

View File

@@ -5,8 +5,9 @@ import "sync/atomic"
// seg is one contiguous byte range of the output file, downloaded by a single
// ranged GET. A segment IS a byte range — a plain HTTP downloader does not need
// the Piece/Segment/block layering that exists only to share code with
// BitTorrent. written is updated with atomic ops so Stat() can read it while a
// worker advances it.
// BitTorrent. start and end are fixed when the file is divided; written is
// advanced (atomically) by the segment's one worker, so Stat() and the resume
// snapshot can read its frontier while it downloads.
type seg struct {
index int
start int64 // first byte offset, inclusive
@@ -15,15 +16,18 @@ type seg struct {
}
func (s *seg) length() int64 { return s.end - s.start + 1 }
func (s *seg) endOff() int64 { return s.end }
func (s *seg) done() bool { return atomic.LoadInt64(&s.written) >= s.length() }
func (s *seg) advance(n int64) { atomic.AddInt64(&s.written, n) }
func (s *seg) addWritten(n int64) { atomic.AddInt64(&s.written, n) }
func (s *seg) progress() int64 { return atomic.LoadInt64(&s.written) }
func (s *seg) offset() int64 { return s.start + atomic.LoadInt64(&s.written) }
func (s *seg) remaining() int64 { return s.length() - atomic.LoadInt64(&s.written) }
// makeSegments divides a file of total bytes into contiguous segments, using at
// most conns of them and never splitting below minSplit. The remainder lands in
// the last segment. With conns==1 (the default) this yields one segment.
// the last segment. With conns==1 (the default) this yields one segment. Each
// segment gets its own worker, so the division here fixes the parallelism: there
// is no later rebalancing.
func makeSegments(total, minSplit int64, conns int) []seg {
if conns < 1 {
conns = 1

View File

@@ -3,6 +3,9 @@ package httpdl
import (
"os"
"path/filepath"
"sort"
"sync"
"sync/atomic"
"testing"
)
@@ -52,14 +55,14 @@ func TestSegProgress(t *testing.T) {
if s.done() {
t.Fatalf("new segment should not be done")
}
s.advance(60)
s.addWritten(60)
if s.offset() != 160 {
t.Errorf("offset = %d, want 160", s.offset())
}
if s.remaining() != 40 {
t.Errorf("remaining = %d, want 40", s.remaining())
}
s.advance(40)
s.addWritten(40)
if !s.done() {
t.Errorf("segment should be done after writing full length")
}
@@ -186,3 +189,107 @@ func TestParseContentRange(t *testing.T) {
t.Errorf("parseContentRange(garbage) ok = true, want false")
}
}
// tilesCover checks that segs cover [0,total) exactly: sorted by start they must
// be contiguous with no gap and no overlap.
func tilesCover(t *testing.T, segs []seg, total int64) {
t.Helper()
sorted := append([]seg(nil), segs...)
sort.Slice(sorted, func(i, j int) bool { return sorted[i].start < sorted[j].start })
var next int64
for _, s := range sorted {
if s.start != next {
t.Fatalf("segment gap/overlap: next byte %d, got start %d", next, s.start)
}
next = s.endOff() + 1
}
if next != total {
t.Fatalf("segments cover %d bytes, want %d", next, total)
}
}
// TestSegmentsConcurrentCoverage drives the segments the way real workers do —
// one worker per segment, each copying its range in small chunks and committing
// with addWritten. The coverage array proves the core invariant: every byte is
// written exactly once, with no gap or overlap between adjacent segments.
func TestSegmentsConcurrentCoverage(t *testing.T) {
const (
total = int64(64 * readBuf) // 2 MiB
minSplit = int64(readBuf)
chunk = int64(readBuf / 4) // one "read"
conns = 8
)
segs := makeSegments(total, minSplit, conns)
if len(segs) < 2 {
t.Fatalf("expected several segments, got %d", len(segs))
}
cover := make([]uint32, total)
var wg sync.WaitGroup
for i := range segs {
wg.Add(1)
go func(s *seg) {
defer wg.Done()
for s.remaining() > 0 {
n := chunk
if s.remaining() < n {
n = s.remaining()
}
off := s.offset()
for j := off; j < off+n; j++ {
atomic.AddUint32(&cover[j], 1)
}
s.addWritten(n)
}
}(&segs[i])
}
wg.Wait()
for i, c := range cover {
if c != 1 {
t.Fatalf("byte %d written %d times, want exactly 1", i, c)
}
}
tilesCover(t, segs, total)
}
// TestSnapshotRoundTrips checks the persistence path an interrupt-and-resume
// relies on: a snapshot of partially-written segments records each frontier, and
// segsFromControl rebuilds a set that still tiles the file and keeps the bytes
// already written.
func TestSnapshotRoundTrips(t *testing.T) {
const (
total = int64(8 * readBuf)
minSplit = int64(readBuf)
)
segs := makeSegments(total, minSplit, 4)
if len(segs) < 2 {
t.Fatalf("expected several segments, got %d", len(segs))
}
// Each segment makes some progress, so the snapshot has a real frontier to
// carry across the round trip.
var want int64
for i := range segs {
n := int64(readBuf)
if n > segs[i].length() {
n = segs[i].length()
}
segs[i].addWritten(n)
want += n
}
c := snapshot("http://x", total, "", "", segs)
if len(c.Segs) != len(segs) {
t.Fatalf("snapshot has %d segments, want %d", len(c.Segs), len(segs))
}
rebuilt := segsFromControl(&c)
tilesCover(t, rebuilt, total)
var written int64
for _, s := range rebuilt {
written += s.progress()
}
if written != want {
t.Errorf("rebuilt written = %d, want %d (bytes must survive the round trip)", written, want)
}
}

115
main.go
View File

@@ -100,9 +100,15 @@ func run(args []string) int {
eng := download.NewEngine(opts.Int("max-concurrent-downloads"))
uiCtx, uiCancel := context.WithCancel(context.Background())
uiDone := make(chan struct{})
if !opts.Bool("quiet") && !opts.Bool("dry-run") {
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)
@@ -129,7 +135,7 @@ func run(args []string) int {
}
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)
@@ -190,9 +196,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}
}
@@ -224,6 +230,24 @@ func dedupeTorrents(uris []string, follow bool) map[string]string {
return dup
}
// markInfoHash records the v1 infohash of the .torrent at path in seen and
// reports whether it repeats one already seen, returning the path first recorded
// under that infohash. Callers fail the duplicate rather than let two jobs share
// one torrent on the client (see bt.SourceInfoHash for why that aliasing is
// unsafe). A v2-only or unreadable infohash is never a duplicate — the dedup
// can't see it, so it runs.
func markInfoHash(seen map[metainfo.Hash]string, path string) (first string, dup bool) {
h, ok := bt.SourceInfoHash(path, true)
if !ok {
return "", false
}
if first, dup = seen[h]; dup {
return first, true
}
seen[h] = path
return "", false
}
func build(opts *cli.Options, targets []target) ([]job, *torrent.Client, error) {
flat := allURIs(targets)
// Two sources naming the same torrent (a magnet and its own .torrent, say)
@@ -263,7 +287,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),
}})
@@ -358,6 +382,7 @@ func runJobs(ctx context.Context, eng *download.Engine, jobs []job) []download.R
// so a regular download that merely ends in ".torrent" is left alone.
func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job {
var out []job
seen := map[metainfo.Hash]string{}
for _, j := range jobs {
h, ok := j.dl.(*httpdl.Download)
if !ok || j.dl.Stat().Status != download.Complete {
@@ -373,6 +398,17 @@ func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job {
if _, err := bt.Files(path); err != nil {
continue // downloaded file is not a valid torrent
}
// Two fetched .torrent files that name the same torrent would share one
// torrent on the refcount-less client: the first to finish Drops it out
// 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{
name: path,
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
}})
continue
}
out = append(out, job{source: path, dl: bt.New(client, path, true, bo)})
}
return out
@@ -408,22 +444,23 @@ func allURIs(ts []target) []string {
// gatherURIs collects download targets from the command line, --torrent-file,
// and --input-file. Plain command-line http/https URIs are grouped into ONE
// mirrored download by default (every command-line URI names the same file);
// -Z/--force-sequential makes each its own download. Torrents and magnets
// are always one target each; --input-file groups TAB-separated URIs per line.
// -Z/--force-sequential makes each its own download. Torrents, magnets, and
// .torrent URLs (fetched over HTTP then followed) are always one target each,
// since distinct torrents are not mirrors of one another; --input-file groups
// TAB-separated URIs per line.
func gatherURIs(opts *cli.Options, cmdURIs []string) []target {
var ts []target
follow := opts.Bool("follow-torrent")
seq := opts.Bool("force-sequential")
var httpGroup []string
for _, u := range cmdURIs {
switch uriKind(u, follow) {
case kindHTTP, kindFollow:
if seq {
ts = append(ts, target{uris: []string{u}})
} else {
// Only plain HTTP URLs collapse into one mirror download (alternate sources
// for the same file); -Z opts out. A .torrent URL (kindFollow), like a
// torrent or magnet, is its own download — distinct torrents are never
// mirrors of one another.
if uriKind(u, follow) == kindHTTP && !seq {
httpGroup = append(httpGroup, u)
}
default:
} else {
ts = append(ts, target{uris: []string{u}})
}
}
@@ -679,42 +716,29 @@ func exitCode(err error) int {
// isNameResolution reports whether err is (or wraps) a DNS lookup failure —
// "name resolution failed" (19), which usually points at the user's own
// network or resolver rather than a dead remote.
// network or resolver rather than a dead remote. The stdlib reports every such
// failure (no-such-host, server-misbehaving) as *net.DNSError.
func isNameResolution(err error) bool {
var de *net.DNSError
if errors.As(err, &de) {
return true
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "no such host") || strings.Contains(msg, "server misbehaving")
return errors.As(err, &de)
}
// isNetwork reports whether err is a connection-level failure — refused, reset,
// or an unreachable host/network — a "network problem" (6).
// or an unreachable host/network — a "network problem" (6). A dial error from
// net/http wraps the syscall errno, so the typed check sees through it.
func isNetwork(err error) bool {
if errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ECONNRESET) ||
errors.Is(err, syscall.ENETUNREACH) || errors.Is(err, syscall.EHOSTUNREACH) {
return true
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "connection refused") ||
strings.Contains(msg, "connection reset") ||
strings.Contains(msg, "network is unreachable") ||
strings.Contains(msg, "no route to host")
return errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ECONNRESET) ||
errors.Is(err, syscall.ENETUNREACH) || errors.Is(err, syscall.EHOSTUNREACH)
}
// isTimeout reports whether err is (or wraps) a timeout: a deadline exceeded, a
// net.Error that timed out, or an idle-read timeout reported as such.
// net.Error that timed out, or our own idle-read timeout (httpdl.ErrTimeout).
func isTimeout(err error) bool {
if errors.Is(err, context.DeadlineExceeded) {
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, httpdl.ErrTimeout) {
return true
}
var ne net.Error
if errors.As(err, &ne) && ne.Timeout() {
return true
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out")
return errors.As(err, &ne) && ne.Timeout()
}
// saveSession writes the sources of every download that did not finish to path,
@@ -799,6 +823,11 @@ func makeLimiter(bps int64) *rate.Limiter {
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.
func parseSelect(s string) map[int]bool {
s = strings.TrimSpace(s)
@@ -816,10 +845,14 @@ func parseSelect(s string) map[int]bool {
continue
}
b, _ := strconv.Atoi(strings.TrimSpace(hi))
for i := a; i <= b; i++ {
if i > 0 {
out[i] = true
if a < 1 {
a = 1
}
if b > maxSelectIndex {
b = maxSelectIndex
}
for i := a; i <= b; i++ {
out[i] = true
}
}
return out

View File

@@ -11,6 +11,9 @@ import (
"syscall"
"testing"
"github.com/anacrolix/torrent/bencode"
"github.com/anacrolix/torrent/metainfo"
"github.com/hanbok/got/bt"
"github.com/hanbok/got/cli"
"github.com/hanbok/got/download"
"github.com/hanbok/got/httpdl"
@@ -66,16 +69,14 @@ func TestExitCode(t *testing.T) {
err error
want int
}{
{"idle timeout", errors.New("idle timeout: no data received"), 2},
{"metadata timed out", errors.New("timed out fetching metadata after 3s"), 2},
{"idle timeout", fmt.Errorf("segment 0: %w (after 1m0s)", httpdl.ErrTimeout), 2},
{"metadata timed out", fmt.Errorf("timed out fetching metadata after 3s: %w", context.DeadlineExceeded), 2},
{"deadline", context.DeadlineExceeded, 2},
{"file exists", fmt.Errorf("/tmp/x: %w (use --allow-overwrite or -c)", httpdl.ErrOutputExists), 13},
{"dns typed", dnsErr, 19},
{"dns wrapped through retries", fmt.Errorf("http://x: %w (after 5 tries)", dnsErr), 19},
{"dns by message", errors.New("dial tcp: lookup nope.invalid: no such host"), 19},
{"refused typed", refused, 6},
{"refused wrapped through retries", fmt.Errorf("http://x: %w (after 5 tries)", refused), 6},
{"refused by message", errors.New("dial tcp 127.0.0.1:1: connect: connection refused"), 6},
{"not found", fmt.Errorf("%w: http://x/y: 404 Not Found", httpdl.ErrNotFound), 3},
{"too slow", fmt.Errorf("http://x: %w (<= %d bytes/sec)", httpdl.ErrTooSlow, 1024), 5},
{"checksum", fmt.Errorf("/tmp/x: %w (want a, got b)", httpdl.ErrChecksum), 32},
@@ -112,8 +113,8 @@ func TestDedupeTorrents(t *testing.T) {
}
// TestGatherURIsGrouping checks the command-line grouping: plain http URIs
// become one mirrored download by default, -Z splits them, and torrents/magnets
// stay separate.
// become one mirrored download by default, -Z splits them, and torrents,
// magnets, and .torrent URLs stay separate.
func TestGatherURIsGrouping(t *testing.T) {
parse := func(args ...string) []target {
res, err := cli.Parse(append([]string{"--no-conf"}, args...))
@@ -131,6 +132,16 @@ func TestGatherURIsGrouping(t *testing.T) {
if ts := parse("http://a/x", "magnet:?xt=urn:btih:abc"); len(ts) != 2 {
t.Errorf("http + magnet: got %d targets, want 2 (magnet never mirrors)", len(ts))
}
// Distinct .torrent URLs are separate downloads, not mirrors of one file
// (follow-torrent defaults on, so each is fetched then followed).
if ts := parse("http://a/x.torrent", "http://b/y.torrent", "http://c/z.torrent"); len(ts) != 3 {
t.Errorf(".torrent URLs: got %d targets, want 3 separate downloads", len(ts))
}
// A plain http URL still mirror-groups even alongside a .torrent URL: one
// http group of 1 + one torrent target = 2.
if ts := parse("http://a/x", "http://b/y.torrent"); len(ts) != 2 {
t.Errorf("http + .torrent: got %d targets, want 2", len(ts))
}
}
// TestReadInputFileTAB checks the input-file grouping: TAB-separated URIs on
@@ -152,10 +163,69 @@ func TestReadInputFileTAB(t *testing.T) {
}
}
// writeTorrent creates a .torrent file at dir/name whose info dict is built from
// a file of the given content and internal name. Two calls with the same
// internalName and content produce the same infohash regardless of the output
// filename, so the test can make two .torrent files that name one torrent.
func writeTorrent(t *testing.T, dir, name, internalName string, content []byte) string {
t.Helper()
data := filepath.Join(dir, internalName)
if err := os.WriteFile(data, content, 0o644); err != nil {
t.Fatal(err)
}
info := metainfo.Info{PieceLength: 32 * 1024}
if err := info.BuildFromFilePath(data); err != nil {
t.Fatal(err)
}
b, err := bencode.Marshal(info)
if err != nil {
t.Fatal(err)
}
tp := filepath.Join(dir, name)
f, err := os.Create(tp)
if err != nil {
t.Fatal(err)
}
defer f.Close()
if err := (&metainfo.MetaInfo{InfoBytes: b}).Write(f); err != nil {
t.Fatal(err)
}
return tp
}
// TestFollowUpInfoHashDedup covers the fix for two fetched .torrent files that
// name the same torrent: the second must be flagged a duplicate so followUps
// fails it rather than letting both share — and Drop — one torrent on the client.
func TestFollowUpInfoHashDedup(t *testing.T) {
dir := t.TempDir()
a := writeTorrent(t, dir, "a.torrent", "same", []byte("AAAAAAAA"))
b := writeTorrent(t, dir, "b.torrent", "same", []byte("AAAAAAAA")) // same infohash as a
c := writeTorrent(t, dir, "c.torrent", "diff", []byte("BBBBBBBB")) // a distinct torrent
// Sanity-check the fixture: a and b really share an infohash, c differs.
ha, _ := bt.SourceInfoHash(a, true)
hb, _ := bt.SourceInfoHash(b, true)
hc, _ := bt.SourceInfoHash(c, true)
if ha != hb || ha == hc {
t.Fatalf("fixture: want a==b!=c infohashes, got a=%v b=%v c=%v", ha, hb, hc)
}
seen := map[metainfo.Hash]string{}
if first, dup := markInfoHash(seen, a); dup {
t.Fatalf("a is the first occurrence, wrongly flagged a duplicate of %s", first)
}
if first, dup := markInfoHash(seen, b); !dup || first != a {
t.Errorf("b should duplicate a: got first=%q dup=%v, want %q true", first, dup, a)
}
if _, dup := markInfoHash(seen, c); dup {
t.Errorf("c is a distinct torrent, wrongly flagged a duplicate")
}
}
func TestReportExit(t *testing.T) {
canceled := download.Result{Name: "a", Err: context.Canceled}
done := download.Result{Name: "b", Err: nil}
timeout := download.Result{Name: "c", Err: errors.New("idle timeout")}
timeout := download.Result{Name: "c", Err: httpdl.ErrTimeout}
tests := []struct {
name string

View File

@@ -25,8 +25,10 @@ type Reporter struct {
w io.Writer
tty bool
width int
height int
win map[string]*speedWindow // keyed by Stat.ID
prevLines int // lines drawn in the previous TTY frame, redrawn in place
lastLog time.Time // last non-TTY line emitted
}
@@ -62,6 +64,7 @@ func New(snap func() []download.Stat, human bool) *Reporter {
w: os.Stdout,
tty: term.IsTerminal(int(os.Stdout.Fd())),
width: 80,
height: 24,
win: map[string]*speedWindow{},
}
}
@@ -103,29 +106,21 @@ func (r *Reporter) render(final bool) {
}
}
var line string
switch len(stats) {
case 0:
if !final && r.tty {
return // nothing active; leave the line as-is
}
case 1:
line = r.lineOne(stats[0])
default:
line = r.lineMany(stats)
}
if r.tty {
r.width = termWidth(r.width)
line = clampRunes(line, r.width)
fmt.Fprint(r.w, "\r"+line+"\x1b[K")
if final {
fmt.Fprintln(r.w)
}
r.height = termHeight(r.height)
r.redraw(r.block(stats), final)
return
}
// Not a TTY: can't redraw one line, so print sparingly (and always the
// final line) to keep redirected output and logs readable.
// Not a TTY: no cursor control, so emit a single line sparingly (and always
// the final one) to keep redirected output and logs readable.
var line string
switch {
case len(stats) == 1:
line = r.lineOne(stats[0])
case len(stats) > 1:
line = r.summary(stats)
}
if line != "" && (final || now.Sub(r.lastLog) >= logEvery) {
fmt.Fprintln(r.w, line)
r.lastLog = now
@@ -158,34 +153,189 @@ func (r *Reporter) lineOne(s download.Stat) string {
fmt.Fprintf(&b, " UL:%s", speed(ul, r.human))
}
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(eta))
fmt.Fprintf(&b, " ETA:%s", secfmt(etaDuration(s.Total-s.Completed, dl)))
}
b.WriteByte(']')
return b.String()
}
func (r *Reporter) lineMany(stats []download.Stat) string {
// nameW is the fixed rune width of the name column in the table, matching
// shortName's cap so a padded name lines the columns up.
const nameW = 20
// block builds the lines of the live display: the single-download dashboard for
// one download, a name/bar/speed/ETA table for several, or a one-line summary
// when the table would not fit the window.
func (r *Reporter) block(stats []download.Stat) []string {
switch {
case len(stats) == 0:
return nil
case len(stats) == 1:
return []string{r.lineOne(stats[0])}
case len(stats)+1 > r.height: // header + one row each would overflow the screen
return []string{r.summary(stats)}
default:
out := make([]string, 0, len(stats)+1)
out = append(out, r.summary(stats))
for _, s := range stats {
out = append(out, r.tableRow(s))
}
return out
}
}
// redraw repaints the block in place: move to the top of the previous frame,
// clear it, and print the new lines clamped to the terminal width. A stray line
// from stderr that lands in the block is painted over on the next tick.
func (r *Reporter) redraw(lines []string, final bool) {
if r.prevLines > 0 {
if r.prevLines > 1 {
fmt.Fprintf(r.w, "\x1b[%dA", r.prevLines-1) // up to the first line
}
fmt.Fprint(r.w, "\r\x1b[J") // column 0, clear to end of screen
}
for i, ln := range lines {
if i > 0 {
fmt.Fprint(r.w, "\n")
}
fmt.Fprint(r.w, clampRunes(ln, r.width))
}
r.prevLines = len(lines)
if final && len(lines) > 0 {
fmt.Fprintln(r.w) // leave the cursor below the block for the OK/FAIL lines
r.prevLines = 0
}
}
// summary is the header/aggregate line: the active count and total down/up
// speed, plus a stalled count when any download has gone quiet.
func (r *Reporter) summary(stats []download.Stat) string {
if len(stats) == 0 {
return ""
}
var totDL, totUL int64
stalled, waiting := 0, 0
for _, s := range stats {
dl, ul := r.rates(s)
totDL += dl
totUL += ul
if s.Status == download.Waiting {
waiting++
continue
}
var b strings.Builder
fmt.Fprintf(&b, "[%d active DL:%s UL:%s]", len(stats), speed(totDL, r.human), speed(totUL, r.human))
for i, s := range stats {
if i >= 4 {
fmt.Fprintf(&b, "[+%d]", len(stats)-i)
break
}
if s.Total > 0 {
fmt.Fprintf(&b, "[%s %d%%]", shortName(s.Name), percent(s.Completed, s.Total))
} else {
fmt.Fprintf(&b, "[%s %s]", shortName(s.Name), humanSize(s.Completed, r.human))
if r.isStalled(s, dl) {
stalled++
}
}
return b.String()
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 {
line += fmt.Sprintf(" (%d stalled)", stalled)
}
return line
}
// tableRow renders one download as "name |bar| pct metric tail": the bar and
// percent for progress, then the field that answers "stalled or fast" (DL speed
// or seeding) and a tail (ETA, upload rate, or a stalled flag).
func (r *Reporter) tableRow(s download.Stat) string {
dl, ul := r.rates(s)
metric, tail := r.rowMetric(s, dl, ul)
return fmt.Sprintf(" %s %s %s %-9s %s", padRune(shortName(s.Name), nameW), barOf(s), pctField(s), metric, tail)
}
// rowMetric returns a table row's two trailing fields for the download's state:
// the primary metric (DL speed, "seeding", "done", "failed") and a tail (ETA,
// upload rate, or a "stalled" flag).
func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string) {
switch s.Status {
case download.Waiting:
return "queued", ""
case download.Seeding:
return "seeding", "UL:" + speed(ul, r.human)
case download.Complete:
return "done", ""
case download.Errored:
return "failed", ""
default:
metric = "DL:" + speed(dl, r.human)
switch {
case dl > 0 && s.Total > 0:
tail = secfmt(etaDuration(s.Total-s.Completed, dl))
case r.isStalled(s, dl):
tail = "stalled"
}
return metric, tail
}
}
// 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,
// 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.
func (r *Reporter) isStalled(s download.Stat, dl int64) bool {
if s.Status != download.Active || dl > 0 {
return false
}
w := r.win[s.ID]
if w == nil || len(w.samples) < 2 {
return false
}
return w.samples[len(w.samples)-1].t.Sub(w.samples[0].t) >= 4*time.Second
}
// barOf renders a fixed-width progress bar. A seeding or complete download is
// full; an unknown total (no Content-Length, or pre-metadata) shows empty.
func barOf(s download.Stat) string {
const cells = 12
k := 0
switch {
case s.Status == download.Seeding || s.Status == download.Complete:
k = cells
case s.Total > 0:
k = int(int64(cells) * s.Completed / s.Total)
if k > cells {
k = cells
}
}
return "|" + strings.Repeat("#", k) + strings.Repeat("-", cells-k) + "|"
}
// pctField is the percent column: a number, full for seeding/complete, or "--"
// when the total is unknown.
func pctField(s download.Stat) string {
switch {
case s.Status == download.Seeding || s.Status == download.Complete:
return "100%"
case s.Total > 0:
return fmt.Sprintf("%3d%%", percent(s.Completed, s.Total))
default:
return " --"
}
}
// padRune right-pads s with spaces to w runes (it is never longer, having been
// through shortName) so table columns line up on rune count.
func padRune(s string, w int) string {
if n := utf8.RuneCountInString(s); n < w {
return s + strings.Repeat(" ", w-n)
}
return s
}
// add records a sample and drops any that have fallen out of the window. A
@@ -272,3 +422,13 @@ func termWidth(prev int) int {
}
return 80
}
func termHeight(prev int) int {
if _, h, err := term.GetSize(int(os.Stdout.Fd())); err == nil && h > 0 {
return h
}
if prev > 0 {
return prev
}
return 24
}

View File

@@ -1,6 +1,7 @@
package progress
import (
"bytes"
"strings"
"testing"
"time"
@@ -112,6 +113,100 @@ func TestLineOneCNSD(t *testing.T) {
}
}
func TestBarOf(t *testing.T) {
tests := []struct {
name string
s download.Stat
want string
}{
{"empty", download.Stat{Status: download.Active, Total: 100, Completed: 0}, "|------------|"},
{"half", download.Stat{Status: download.Active, Total: 100, Completed: 50}, "|######------|"},
{"full by complete", download.Stat{Status: download.Complete}, "|############|"},
{"full by seeding", download.Stat{Status: download.Seeding}, "|############|"},
{"unknown total is empty", download.Stat{Status: download.Active, Total: -1, Completed: 9}, "|------------|"},
}
for _, tc := range tests {
if got := barOf(tc.s); got != tc.want {
t.Errorf("%s: barOf = %q, want %q", tc.name, got, tc.want)
}
}
}
func TestPctField(t *testing.T) {
if got := pctField(download.Stat{Status: download.Active, Total: 100, Completed: 50}); got != " 50%" {
t.Errorf("active 50%% = %q, want ' 50%%'", got)
}
if got := pctField(download.Stat{Status: download.Seeding}); got != "100%" {
t.Errorf("seeding = %q, want 100%%", got)
}
if got := pctField(download.Stat{Status: download.Active, Total: -1}); got != " --" {
t.Errorf("unknown total = %q, want ' --'", got)
}
}
// TestRowMetric: the two trailing fields per download state.
func TestRowMetric(t *testing.T) {
r := newReporter(true)
if m, tail := r.rowMetric(download.Stat{Status: download.Active, Total: 100, Completed: 50}, 10, 0); !strings.HasPrefix(m, "DL:") || tail == "" {
t.Errorf("active downloading: metric=%q tail=%q, want DL:… + an ETA", m, tail)
}
if m, tail := r.rowMetric(download.Stat{Status: download.Seeding}, 0, 128<<10); m != "seeding" || !strings.HasPrefix(tail, "UL:") {
t.Errorf("seeding: metric=%q tail=%q, want seeding + UL:…", m, tail)
}
if m, _ := r.rowMetric(download.Stat{Status: download.Complete}, 0, 0); m != "done" {
t.Errorf("complete metric=%q, want done", m)
}
}
// TestBlockTableAndFallback: several downloads render as header+rows when they
// fit, and collapse to a single summary line when the window is too short.
func TestBlockTableAndFallback(t *testing.T) {
r := newReporter(true)
stats := []download.Stat{
{ID: "a", Name: "a", Status: download.Active, Total: 100, Completed: 50},
{ID: "b", Name: "b", Status: download.Active, Total: 100, Completed: 10},
}
r.height = 24
if got := r.block(stats); len(got) != 3 { // header + 2 rows
t.Errorf("table block = %d lines, want 3:\n%v", len(got), got)
}
r.height = 2 // header + 2 rows = 3 > 2
if got := r.block(stats); len(got) != 1 {
t.Errorf("overflow block = %d lines, want 1 summary:\n%v", len(got), got)
}
}
// TestRenderRedraw: the TTY path draws the block, then on the next frame moves
// the cursor up over the previous frame and clears it before repainting.
func TestRenderRedraw(t *testing.T) {
snap := []download.Stat{
{ID: "a", Name: "alpha", IsBT: true, Status: download.Active, Total: 100, Completed: 50},
{ID: "b", Name: "beta", IsBT: true, Status: download.Seeding, Total: 100, Completed: 100, Uploaded: 200},
}
var buf bytes.Buffer
r := &Reporter{
snap: func() []download.Stat { return snap },
human: true, w: &buf, tty: true, width: 200, height: 24,
win: map[string]*speedWindow{},
}
r.render(false)
first := buf.String()
for _, want := range []string{"2 active", "alpha", "seeding", "|"} {
if !strings.Contains(first, want) {
t.Fatalf("first frame missing %q:\n%q", want, first)
}
}
buf.Reset()
r.render(false)
second := buf.String()
if !strings.Contains(second, "\x1b[2A") { // up prevLines-1 = 3-1
t.Errorf("second frame should move cursor up 2 lines:\n%q", second)
}
if !strings.Contains(second, "\x1b[J") {
t.Errorf("second frame should clear to end of screen:\n%q", second)
}
}
// TestShortNameRune ensures CJK names are truncated on runes, not bytes.
func TestShortNameRune(t *testing.T) {
name := strings.Repeat("あ", 30) // 30 runes, 90 bytes