Add resumable downloads, download paths, queue controls and history

internal/fetch/copy.go

@@ -0,0 +1,55 @@
+package fetch
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	bufferSize   = 1 << 20
+	stallTimeout = 2 * time.Minute
+)
+
+func copyBody(destination io.Writer, body io.ReadCloser, remaining int64, record func(int64) error) error {
+	var stalled atomic.Bool
+	stallTimer := time.AfterFunc(stallTimeout, func() {
+		stalled.Store(true)
+		body.Close()
+	})
+	defer stallTimer.Stop()
+
+	buffer := make([]byte, bufferSize)
+	for remaining != 0 {
+		readSize := len(buffer)
+		if remaining > 0 && remaining < int64(readSize) {
+			readSize = int(remaining)
+		}
+
+		count, readErr := body.Read(buffer[:readSize])
+		if count > 0 {
+			stallTimer.Reset(stallTimeout)
+			if _, err := destination.Write(buffer[:count]); err != nil {
+				return err
+			}
+			if err := record(int64(count)); err != nil {
+				return err
+			}
+			if remaining > 0 {
+				remaining -= int64(count)
+			}
+		}
+		if errors.Is(readErr, io.EOF) {
+			return nil
+		}
+		if readErr != nil {
+			if stalled.Load() {
+				return ErrStalled
+			}
+			return fmt.Errorf("%w: %w", errConnectionFailed, readErr)
+		}
+	}
+	return nil
+}

internal/fetch/downloader.go

@@ -3,25 +3,15 @@ package fetch
 import (
 	"context"
 	"errors"
-	"fmt"
-	"io"
 	"net/http"
 	"os"
 	"path/filepath"
-	"time"
 
 	"ddc/internal/diskspace"
 	"ddc/internal/files"
 	"ddc/internal/guard"
 )
 
-const (
-	userAgent          = "ddc/1.0"
-	bufferSize         = 1 << 20
-	stallTimeout       = 2 * time.Minute
-	spaceCheckInterval = int64(256) << 20
-)
-
 var (
 	ErrIncomplete = errors.New("download ended before the full file was received")
 	ErrStalled    = errors.New("download stalled, no data received for too long")
@@ -29,8 +19,21 @@ var (
 
 type Progress interface {
 	SetTotal(total int64)
+	SetDone(done int64)
 	Add(delta int64)
 	SetFileName(name string)
+	SetNotice(text string)
+}
+
+type Request struct {
+	URL           string
+	Directory     string
+	FileName      string
+	ReservedBytes int64
+	Connections   int
+	State         *State
+	Progress      Progress
+	Checkpoint    func(State)
 }
 
 type Downloader struct {
@@ -42,111 +45,99 @@ func NewDownloader(policy *guard.Policy) *Downloader {
 	return &Downloader{policy: policy, client: policy.NewHTTPClient()}
 }
 
-func (downloader *Downloader) Download(ctx context.Context, rawURL, directory string, reservedBytes int64, progress Progress) (string, error) {
-	target, err := downloader.policy.ValidateURL(rawURL)
+func (downloader *Downloader) Download(ctx context.Context, request Request) (string, error) {
+	target, err := downloader.policy.ValidateURL(request.URL)
 	if err != nil {
 		return "", err
 	}
 
-	ctx, cancel := context.WithCancelCause(ctx)
-	defer cancel(nil)
-
-	response, err := downloader.open(ctx, target.String())
+	source := &remoteFile{url: target.String()}
+	partPath := filepath.Join(request.Directory, files.PartFileName)
+	state, probe, err := downloader.prepare(ctx, source, request, partPath)
 	if err != nil {
 		return "", err
 	}
-	defer response.Body.Close()
 
-	fileName := FileNameFromResponse(response)
-	progress.SetFileName(fileName)
-	if err := diskspace.EnsureFree(directory, max(response.ContentLength, 0), reservedBytes); err != nil {
+	progress := request.Progress
+	progress.SetFileName(state.FileName)
+	progress.SetTotal(state.TotalBytes)
+	progress.SetDone(state.DoneBytes())
+	remainingBytes := max(state.TotalBytes-state.DoneBytes(), 0)
+	if err := diskspace.EnsureFree(request.Directory, remainingBytes, request.ReservedBytes); err != nil {
+		closeBody(probe)
 		return "", err
 	}
-	progress.SetTotal(max(response.ContentLength, 0))
 
-	partPath := filepath.Join(directory, files.PartFileName)
-	written, err := writePart(cancel, partPath, response.Body, progress, directory, reservedBytes)
+	file, err := openPartFile(partPath, probe != nil)
 	if err != nil {
-		os.Remove(partPath)
-		if errors.Is(context.Cause(ctx), ErrStalled) {
-			return "", ErrStalled
-		}
+		closeBody(probe)
 		return "", err
 	}
-	if response.ContentLength > 0 && written != response.ContentLength {
-		os.Remove(partPath)
-		return "", ErrIncomplete
+	runErr := newSession(downloader, source, file, request, state).run(ctx, probe)
+	closeErr := file.Close()
+	if runErr != nil {
+		return "", runErr
+	}
+	if closeErr != nil {
+		return "", closeErr
 	}
 
-	finalPath := files.UniquePath(filepath.Join(directory, fileName))
+	finalPath := files.UniquePath(filepath.Join(request.Directory, state.FileName))
 	if err := os.Rename(partPath, finalPath); err != nil {
-		os.Remove(partPath)
 		return "", err
 	}
 	return finalPath, nil
 }
 
-func (downloader *Downloader) open(ctx context.Context, target string) (*http.Response, error) {
-	request, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
-	if err != nil {
-		return nil, err
+func (downloader *Downloader) prepare(ctx context.Context, source *remoteFile, request Request, partPath string) (State, *http.Response, error) {
+	if request.State != nil && request.State.valid() && fileExists(partPath) {
+		state := *request.State
+		source.validator = state.Validator
+		source.totalBytes = state.TotalBytes
+		return state, nil, nil
 	}
-	request.Header.Set("User-Agent", userAgent)
-	request.Header.Set("Accept-Encoding", "identity")
 
-	response, err := downloader.client.Do(request)
+	probe, err := downloader.open(ctx, source, 0, 0)
 	if err != nil {
-		return nil, err
+		return State{}, nil, err
 	}
-	if response.StatusCode != http.StatusOK {
+	source.describe(probe)
+
+	connections := 1
+	if probe.Header.Get("Accept-Ranges") == "bytes" {
+		connections = min(max(request.Connections, 1), MaxConnections)
+	}
+	state := State{
+		FileName:   chooseFileName(request.FileName, probe),
+		Validator:  source.validator,
+		TotalBytes: source.totalBytes,
+		Segments:   planSegments(source.totalBytes, connections),
+	}
+	return state, probe, nil
+}
+
+func chooseFileName(requested string, response *http.Response) string {
+	if name := SanitizeFileName(requested); name != "" {
+		return name
+	}
+	return FileNameFromResponse(response)
+}
+
+func openPartFile(partPath string, fresh bool) (*os.File, error) {
+	flags := os.O_WRONLY | os.O_CREATE
+	if fresh {
+		flags |= os.O_TRUNC
+	}
+	return os.OpenFile(partPath, flags, 0o644)
+}
+
+func fileExists(path string) bool {
+	info, err := os.Stat(path)
+	return err == nil && info.Mode().IsRegular()
+}
+
+func closeBody(response *http.Response) {
+	if response != nil {
 		response.Body.Close()
-		return nil, fmt.Errorf("server responded with %s", response.Status)
-	}
-	return response, nil
-}
-
-func writePart(cancel context.CancelCauseFunc, partPath string, body io.Reader, progress Progress, directory string, reservedBytes int64) (int64, error) {
-	file, err := os.OpenFile(partPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
-	if err != nil {
-		return 0, err
-	}
-	written, copyErr := copyBody(cancel, file, body, progress, directory, reservedBytes)
-	closeErr := file.Close()
-	if copyErr != nil {
-		return written, copyErr
-	}
-	return written, closeErr
-}
-
-func copyBody(cancel context.CancelCauseFunc, destination io.Writer, body io.Reader, progress Progress, directory string, reservedBytes int64) (int64, error) {
-	stallTimer := time.AfterFunc(stallTimeout, func() { cancel(ErrStalled) })
-	defer stallTimer.Stop()
-
-	buffer := make([]byte, bufferSize)
-	var written, sinceSpaceCheck int64
-	for {
-		count, readErr := body.Read(buffer)
-		if count > 0 {
-			stallTimer.Reset(stallTimeout)
-			if _, err := destination.Write(buffer[:count]); err != nil {
-				return written, err
-			}
-			written += int64(count)
-			sinceSpaceCheck += int64(count)
-			progress.Add(int64(count))
-
-			if sinceSpaceCheck >= spaceCheckInterval {
-				sinceSpaceCheck = 0
-				if err := diskspace.EnsureFree(directory, 0, reservedBytes); err != nil {
-					return written, err
-				}
-			}
-		}
-		if errors.Is(readErr, io.EOF) {
-			return written, nil
-		}
-		if readErr != nil {
-			return written, readErr
-		}
 	}
 }

internal/fetch/request.go

@@ -0,0 +1,141 @@
+package fetch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"ddc/internal/guard"
+)
+
+const userAgent = "ddc/1.0"
+
+var (
+	ErrResumeUnsupported = errors.New("connection was lost and the server does not support resuming")
+	ErrSourceChanged     = errors.New("file changed on the server while downloading")
+	errConnectionFailed  = errors.New("connection to the server failed")
+	errServerUnavailable = errors.New("server is temporarily unavailable")
+)
+
+type remoteFile struct {
+	url        string
+	validator  string
+	totalBytes int64
+}
+
+func (source *remoteFile) describe(response *http.Response) {
+	source.totalBytes = max(response.ContentLength, 0)
+	source.validator = validatorOf(response.Header)
+}
+
+func validatorOf(header http.Header) string {
+	if entityTag := header.Get("ETag"); entityTag != "" && !strings.HasPrefix(entityTag, "W/") {
+		return entityTag
+	}
+	return header.Get("Last-Modified")
+}
+
+type cancelingBody struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+}
+
+func (body *cancelingBody) Close() error {
+	body.cancel()
+	return body.ReadCloser.Close()
+}
+
+func (downloader *Downloader) open(ctx context.Context, source *remoteFile, position, end int64) (*http.Response, error) {
+	requestContext, cancelRequest := context.WithCancel(ctx)
+	response, err := downloader.send(requestContext, source, position, end)
+	if err != nil {
+		cancelRequest()
+		return nil, err
+	}
+	response.Body = &cancelingBody{ReadCloser: response.Body, cancel: cancelRequest}
+	return response, nil
+}
+
+func (downloader *Downloader) send(ctx context.Context, source *remoteFile, position, end int64) (*http.Response, error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, source.url, nil)
+	if err != nil {
+		return nil, err
+	}
+	request.Header.Set("User-Agent", userAgent)
+	request.Header.Set("Accept-Encoding", "identity")
+
+	ranged := position > 0 || (end > 0 && end < source.totalBytes)
+	if ranged {
+		request.Header.Set("Range", rangeHeader(position, end))
+		if source.validator != "" {
+			request.Header.Set("If-Range", source.validator)
+		}
+	}
+
+	response, err := downloader.client.Do(request)
+	if err != nil {
+		return nil, connectionError(err)
+	}
+	if err := checkResponse(response, source, position, ranged); err != nil {
+		response.Body.Close()
+		return nil, err
+	}
+	return response, nil
+}
+
+func rangeHeader(position, end int64) string {
+	if end > 0 {
+		return fmt.Sprintf("bytes=%d-%d", position, end-1)
+	}
+	return fmt.Sprintf("bytes=%d-", position)
+}
+
+func checkResponse(response *http.Response, source *remoteFile, position int64, ranged bool) error {
+	status := response.StatusCode
+	switch {
+	case !ranged && status == http.StatusOK:
+		if source.totalBytes > 0 && response.ContentLength != source.totalBytes {
+			return ErrSourceChanged
+		}
+		return nil
+	case ranged && status == http.StatusPartialContent:
+		return checkContentRange(response.Header.Get("Content-Range"), source.totalBytes, position)
+	case ranged && status == http.StatusOK:
+		if source.validator != "" && validatorOf(response.Header) != source.validator {
+			return ErrSourceChanged
+		}
+		return ErrResumeUnsupported
+	case status == http.StatusTooManyRequests || status >= http.StatusInternalServerError:
+		return fmt.Errorf("%w (%s)", errServerUnavailable, response.Status)
+	default:
+		return fmt.Errorf("server responded with %s", response.Status)
+	}
+}
+
+func checkContentRange(header string, totalBytes, offset int64) error {
+	rangeText, found := strings.CutPrefix(header, "bytes ")
+	if !found {
+		return ErrResumeUnsupported
+	}
+	span, totalText, found := strings.Cut(rangeText, "/")
+	startText, _, _ := strings.Cut(span, "-")
+	start, err := strconv.ParseInt(startText, 10, 64)
+	if !found || err != nil || start != offset {
+		return ErrResumeUnsupported
+	}
+	if totalBytes > 0 && totalText != "*" && totalText != strconv.FormatInt(totalBytes, 10) {
+		return ErrSourceChanged
+	}
+	return nil
+}
+
+func connectionError(err error) error {
+	if errors.Is(err, guard.ErrBlockedAddress) || errors.Is(err, guard.ErrBlockedPort) || errors.Is(err, guard.ErrTooManyRedirects) {
+		return err
+	}
+	return fmt.Errorf("%w: %w", errConnectionFailed, err)
+}

internal/fetch/session.go

@@ -0,0 +1,128 @@
+package fetch
+
+import (
+	"context"
+	"net/http"
+	"os"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"ddc/internal/diskspace"
+)
+
+const (
+	checkpointInterval = 10 * time.Second
+	spaceCheckInterval = int64(256) << 20
+)
+
+type session struct {
+	downloader      *Downloader
+	source          *remoteFile
+	file            *os.File
+	request         Request
+	state           State
+	segments        []*liveSegment
+	sinceSpaceCheck atomic.Int64
+}
+
+func newSession(downloader *Downloader, source *remoteFile, file *os.File, request Request, state State) *session {
+	segments := make([]*liveSegment, len(state.Segments))
+	for index, segment := range state.Segments {
+		segments[index] = newLiveSegment(segment)
+	}
+	return &session{
+		downloader: downloader,
+		source:     source,
+		file:       file,
+		request:    request,
+		state:      state,
+		segments:   segments,
+	}
+}
+
+func (session *session) run(ctx context.Context, probe *http.Response) error {
+	session.saveCheckpoint()
+	stopSaving := make(chan struct{})
+	savingStopped := make(chan struct{})
+	go session.saveCheckpoints(stopSaving, savingStopped)
+
+	err := session.runWorkers(ctx, probe)
+	close(stopSaving)
+	<-savingStopped
+	session.saveCheckpoint()
+	return err
+}
+
+func (session *session) runWorkers(ctx context.Context, probe *http.Response) error {
+	ctx, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+
+	var workers sync.WaitGroup
+	var firstFailure sync.Once
+	var failure error
+	for index, segment := range session.segments {
+		if segment.complete() {
+			continue
+		}
+		var initial *http.Response
+		if index == 0 {
+			initial, probe = probe, nil
+		}
+		worker := &segmentWorker{session: session, segment: segment}
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			if err := worker.run(ctx, initial); err != nil {
+				firstFailure.Do(func() {
+					failure = err
+					cancel(err)
+				})
+			}
+		}()
+	}
+	closeBody(probe)
+	workers.Wait()
+	return failure
+}
+
+func (session *session) record(count int64) error {
+	session.request.Progress.Add(count)
+	if session.sinceSpaceCheck.Add(count) < spaceCheckInterval {
+		return nil
+	}
+	session.sinceSpaceCheck.Store(0)
+	return diskspace.EnsureFree(session.request.Directory, 0, session.request.ReservedBytes)
+}
+
+func (session *session) snapshot() State {
+	state := session.state
+	state.Segments = make([]Segment, len(session.segments))
+	for index, segment := range session.segments {
+		state.Segments[index] = segment.snapshot()
+	}
+	return state
+}
+
+func (session *session) saveCheckpoints(stop <-chan struct{}, stopped chan<- struct{}) {
+	defer close(stopped)
+	ticker := time.NewTicker(checkpointInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case <-ticker.C:
+			session.saveCheckpoint()
+		}
+	}
+}
+
+func (session *session) saveCheckpoint() {
+	if session.request.Checkpoint == nil {
+		return
+	}
+	state := session.snapshot()
+	session.file.Sync()
+	session.request.Checkpoint(state)
+}

internal/fetch/state.go

@@ -0,0 +1,97 @@
+package fetch
+
+import "sync/atomic"
+
+const (
+	MaxConnections  = 16
+	minSegmentBytes = int64(32) << 20
+)
+
+type Segment struct {
+	Start int64 `json:"start"`
+	End   int64 `json:"end"`
+	Done  int64 `json:"done"`
+}
+
+type State struct {
+	FileName   string    `json:"fileName"`
+	Validator  string    `json:"validator"`
+	TotalBytes int64     `json:"totalBytes"`
+	Segments   []Segment `json:"segments"`
+}
+
+func (state State) DoneBytes() int64 {
+	var done int64
+	for _, segment := range state.Segments {
+		done += segment.Done
+	}
+	return done
+}
+
+func (state State) valid() bool {
+	if state.FileName == "" || len(state.Segments) == 0 {
+		return false
+	}
+	for _, segment := range state.Segments {
+		if segment.Start < 0 || segment.Done < 0 {
+			return false
+		}
+		if segment.End == 0 && state.TotalBytes > 0 {
+			return false
+		}
+		if segment.End > 0 && segment.Start+segment.Done > segment.End {
+			return false
+		}
+	}
+	return true
+}
+
+func planSegments(totalBytes int64, connections int) []Segment {
+	if totalBytes <= 0 {
+		return []Segment{{}}
+	}
+	count := min(int64(max(connections, 1)), max(totalBytes/minSegmentBytes, 1))
+	size := totalBytes / count
+	segments := make([]Segment, count)
+	for index := range segments {
+		start := int64(index) * size
+		segments[index] = Segment{Start: start, End: start + size}
+	}
+	segments[count-1].End = totalBytes
+	return segments
+}
+
+type liveSegment struct {
+	start int64
+	end   int64
+	done  atomic.Int64
+}
+
+func newLiveSegment(segment Segment) *liveSegment {
+	live := &liveSegment{start: segment.Start, end: segment.End}
+	live.done.Store(segment.Done)
+	return live
+}
+
+func (segment *liveSegment) position() int64 {
+	return segment.start + segment.done.Load()
+}
+
+func (segment *liveSegment) bounded() bool {
+	return segment.end > 0
+}
+
+func (segment *liveSegment) remaining() int64 {
+	if !segment.bounded() {
+		return -1
+	}
+	return segment.end - segment.position()
+}
+
+func (segment *liveSegment) complete() bool {
+	return segment.bounded() && segment.position() >= segment.end
+}
+
+func (segment *liveSegment) snapshot() Segment {
+	return Segment{Start: segment.start, End: segment.end, Done: segment.done.Load()}
+}

internal/fetch/worker.go

@@ -0,0 +1,107 @@
+package fetch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+const (
+	maxRetriesWithoutProgress = 8
+	retryBaseDelay            = 2 * time.Second
+	retryMaxDelay             = 30 * time.Second
+)
+
+type segmentWorker struct {
+	session        *session
+	segment        *liveSegment
+	failedAttempts int
+}
+
+func (worker *segmentWorker) run(ctx context.Context, response *http.Response) error {
+	var err error
+	if response == nil {
+		response, err = worker.open(ctx)
+	}
+	for {
+		if err != nil {
+			response, err = worker.reconnect(ctx, err)
+			if err != nil {
+				return err
+			}
+		}
+
+		doneBefore := worker.segment.done.Load()
+		err = worker.copyFrom(response)
+		if err == nil {
+			return nil
+		}
+		if worker.segment.done.Load() > doneBefore {
+			worker.failedAttempts = 0
+		}
+	}
+}
+
+func (worker *segmentWorker) open(ctx context.Context) (*http.Response, error) {
+	return worker.session.downloader.open(ctx, worker.session.source, worker.segment.position(), worker.segment.end)
+}
+
+func (worker *segmentWorker) copyFrom(response *http.Response) error {
+	defer response.Body.Close()
+	writer := io.NewOffsetWriter(worker.session.file, worker.segment.position())
+	if err := copyBody(writer, response.Body, worker.segment.remaining(), worker.record); err != nil {
+		return err
+	}
+	if worker.segment.bounded() && !worker.segment.complete() {
+		return ErrIncomplete
+	}
+	return nil
+}
+
+func (worker *segmentWorker) record(count int64) error {
+	worker.segment.done.Add(count)
+	return worker.session.record(count)
+}
+
+func (worker *segmentWorker) reconnect(ctx context.Context, cause error) (*http.Response, error) {
+	progress := worker.session.request.Progress
+	for {
+		worker.failedAttempts++
+		if ctx.Err() != nil || !isRetryable(cause) || worker.failedAttempts > maxRetriesWithoutProgress {
+			return nil, cause
+		}
+		progress.SetNotice(fmt.Sprintf("Connection lost, reconnecting (attempt %d of %d)", worker.failedAttempts, maxRetriesWithoutProgress))
+		if err := waitBeforeRetry(ctx, worker.failedAttempts); err != nil {
+			return nil, err
+		}
+
+		response, err := worker.open(ctx)
+		if err == nil {
+			progress.SetNotice("")
+			return response, nil
+		}
+		cause = err
+	}
+}
+
+func isRetryable(err error) bool {
+	return errors.Is(err, errConnectionFailed) ||
+		errors.Is(err, errServerUnavailable) ||
+		errors.Is(err, ErrStalled) ||
+		errors.Is(err, ErrIncomplete)
+}
+
+func waitBeforeRetry(ctx context.Context, attempt int) error {
+	delay := min(retryBaseDelay*time.Duration(attempt), retryMaxDelay)
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}