124 lines
3.2 KiB
Go
124 lines
3.2 KiB
Go
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
|
|
logger := worker.session.request.logger().With("part_start", worker.segment.start)
|
|
for {
|
|
worker.failedAttempts++
|
|
if ctx.Err() != nil {
|
|
return nil, cause
|
|
}
|
|
if !isRetryable(cause) {
|
|
logger.Debug("error cannot be retried", "error", cause)
|
|
return nil, cause
|
|
}
|
|
if worker.failedAttempts > maxRetriesWithoutProgress {
|
|
logger.Error("giving up after repeated connection failures", "attempts", maxRetriesWithoutProgress, "error", cause)
|
|
return nil, fmt.Errorf("gave up after %d reconnect attempts: %w", maxRetriesWithoutProgress, cause)
|
|
}
|
|
|
|
logger.Warn("connection lost, reconnecting",
|
|
"attempt", worker.failedAttempts,
|
|
"position", worker.segment.position(),
|
|
"error", 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 {
|
|
logger.Info("reconnected", "position", worker.segment.position())
|
|
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
|
|
}
|
|
}
|