diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index b626db0d..1853f81a 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -274,6 +274,7 @@ type Fetcher struct { prefetchDone atomic.Bool // Prefetch completed or stopped prefetchErr error // Error during prefetch (if any) prefetchStopCh chan struct{} // Signal to stop prefetch + resolveFallback atomic.Bool // Resolve prefetch is still available until the first Range request takes over // Target file file *os.File @@ -467,6 +468,7 @@ func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { // For non-range resources, the response will be used directly in Start if res.Range && res.Size > 0 { f.prefetchStopCh = make(chan struct{}) + f.resolveFallback.Store(true) go f.asyncPrefetch() } @@ -606,6 +608,7 @@ func (f *Fetcher) cleanupPrefetchFile() { // Pause may discard that file before the first Start, so retaining the // count would make the next Start skip bytes that were never copied. f.prefetchSize.Store(0) + f.resolveFallback.Store(false) } func (f *Fetcher) Start() error { @@ -696,17 +699,26 @@ func (f *Fetcher) doStart() error { // For non-range resources, the response will be used directly var prefetchedBytes int64 if f.meta.Res.Range { - // Stop async prefetch and copy data to target file - prefetchedBytes = f.stopPrefetchAndCopyData() + if f.resolveFallback.Load() && !(f.prefetchDone.Load() && f.prefetchErr == nil && f.prefetchSize.Load() >= f.meta.Res.Size) { + // Keep Resolve downloading until the first Range response has been + // validated. The Range request starts at a stable snapshot; a valid + // takeover discards any overlap downloaded by Resolve in the meantime. + prefetchedBytes = f.prefetchSize.Load() + } else { + // Resolve already completed, or no live Resolve fallback remains. + prefetchedBytes = f.stopPrefetchAndCopyData() + } f.resolveDataPos.Store(prefetchedBytes) - // Also close resolve response if still open - f.resolveRespLock.Lock() - if f.resolveResp != nil { - f.resolveResp.Body.Close() - f.resolveResp = nil + if !f.resolveFallback.Load() { + // Also close resolve response if still open. + f.resolveRespLock.Lock() + if f.resolveResp != nil { + f.resolveResp.Body.Close() + f.resolveResp = nil + } + f.resolveRespLock.Unlock() } - f.resolveRespLock.Unlock() } // Avoid request extra modified by extension @@ -722,16 +734,17 @@ func (f *Fetcher) doStart() error { f.ctx, f.cancel = context.WithCancel(context.Background()) // Create downloadLoop lifecycle channel - f.downloadLoopDone = make(chan struct{}) + downloadLoopDone := make(chan struct{}) + f.downloadLoopDone = downloadLoopDone // Start download f.setState(stateSlowStart) - go f.downloadLoop() + go f.downloadLoop(downloadLoopDone) return nil } -func (f *Fetcher) downloadLoop() { +func (f *Fetcher) downloadLoop(done chan struct{}) { ctx := f.ctx defer func() { @@ -741,8 +754,8 @@ func (f *Fetcher) downloadLoop() { } // Signal that downloadLoop has exited - if f.downloadLoopDone != nil { - close(f.downloadLoopDone) + if done != nil { + close(done) } }() @@ -971,7 +984,7 @@ func (f *Fetcher) runConnection(conn *connection) { // response-ready signal was intentionally not used to expand. f.connMu.Lock() conn.exited = true - sequentialRecovery := !f.meta.Res.Range && f.rangeReprobeEligible && len(f.connections) == 1 + sequentialRecovery := !f.meta.Res.Range && (f.rangeReprobeEligible || conn.Completed) && len(f.connections) == 1 f.connMu.Unlock() f.wg.Done() if sequentialRecovery && f.slowStart != nil { @@ -1025,6 +1038,11 @@ func (f *Fetcher) runConnection(conn *connection) { if errors.Is(err, context.Canceled) { return } + requestErr := extractRequestError(err) + shouldUseResolveFallback := requestErr == nil || !isFailureExemptHTTPCode(requestErr.Code) + if shouldUseResolveFallback && f.completeFromResolveFallback(conn) { + return + } if isTerminalRangeError(err) { f.connMu.Lock() conn.lastErr = err @@ -1114,7 +1132,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ if resumeProbe { rangeEnd = f.meta.Res.Size - 1 ifRange = f.ifRange - } else if rangeMode && f.rangeValidatorPinned { + } else if rangeMode && (f.rangeValidatorPinned || f.resolveFallback.Load()) { ifRange = f.ifRange } f.connMu.Unlock() @@ -1169,7 +1187,14 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } expectedResponseLength := int64(-1) + var responseBytesToDiscard int64 if rangeRequested && resp.StatusCode == base.HttpCodeOK { + if conn.ID == 0 && f.resolveFallback.CompareAndSwap(true, false) { + // This response is itself a complete representation, so the older + // Resolve stream is no longer needed. Stop and copy its prefix before + // the sequential restart truncates/overwrites the target from zero. + f.stopPrefetchAndCopyData() + } if resumeProbe { // If-Range returning 200 provides the complete current representation. // Discard the old prefix counters and consume this response from byte 0. @@ -1189,6 +1214,11 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ resp.Body.Close() return err } + responseBytesToDiscard, err = f.commitResolveTakeover(conn, rangeStart) + if err != nil { + resp.Body.Close() + return err + } if resumeProbe { // The origin recovered Range support. Convert the preserved sequential // prefix into a normal ranged chunk before writing the suffix. @@ -1264,59 +1294,68 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ return fmt.Errorf("%w: response body exceeds expected length %d", errInvalidRangeResponse, expectedResponseLength) } + data := buf[:n] + if responseBytesToDiscard > 0 { + discard := min(int64(len(data)), responseBytesToDiscard) + data = data[discard:] + responseBytesToDiscard -= discard + } + finished := false var writeOffset int64 - // Lock to safely read chunk state and calculate write parameters - // This protects against concurrent chunk splitting by helpOtherConnection - f.connMu.Lock() - if f.meta.Res.Range { - // Check current chunk boundaries - this respects any concurrent chunk splitting - remain := conn.Chunk.remain() - if remain <= 0 { - // Chunk has been fully downloaded (possibly split and reduced) - f.connMu.Unlock() + if len(data) > 0 { + // Lock to safely read chunk state and calculate write parameters + // This protects against concurrent chunk splitting by helpOtherConnection + f.connMu.Lock() + if f.meta.Res.Range { + // Check current chunk boundaries - this respects any concurrent chunk splitting + remain := conn.Chunk.remain() + if remain <= 0 { + // Chunk has been fully downloaded (possibly split and reduced) + f.connMu.Unlock() + return nil + } + if remain < int64(len(data)) { + data = data[:remain] + finished = true + } + } + writeOffset = conn.Chunk.Begin + conn.Chunk.Downloaded + f.connMu.Unlock() + + f.fileMu.Lock() + if f.file != nil { + _, writeErr := f.file.WriteAt(data, writeOffset) + if writeErr != nil { + f.fileMu.Unlock() + return writeErr + } + } + f.fileMu.Unlock() + + // Lock again to update Downloaded atomically with the read above + f.connMu.Lock() + conn.Chunk.Downloaded += int64(len(data)) + conn.Downloaded += int64(len(data)) + // Update connection speed periodically + now := time.Now().UnixNano() + if conn.lastSpeedCheck == 0 { + conn.lastSpeedCheck = now + conn.lastSpeedDownload = conn.Downloaded + } else if now-conn.lastSpeedCheck >= int64(500*time.Millisecond) { + elapsed := float64(now-conn.lastSpeedCheck) / float64(time.Second) + if elapsed > 0 { + conn.speed = int64(float64(conn.Downloaded-conn.lastSpeedDownload) / elapsed) + } + conn.lastSpeedCheck = now + conn.lastSpeedDownload = conn.Downloaded + } + f.connMu.Unlock() + + if finished { return nil } - if remain < int64(n) { - n = int(remain) - finished = true - } - } - writeOffset = conn.Chunk.Begin + conn.Chunk.Downloaded - f.connMu.Unlock() - - f.fileMu.Lock() - if f.file != nil { - _, writeErr := f.file.WriteAt(buf[:n], writeOffset) - if writeErr != nil { - f.fileMu.Unlock() - return writeErr - } - } - f.fileMu.Unlock() - - // Lock again to update Downloaded atomically with the read above - f.connMu.Lock() - conn.Chunk.Downloaded += int64(n) - conn.Downloaded += int64(n) - // Update connection speed periodically - now := time.Now().UnixNano() - if conn.lastSpeedCheck == 0 { - conn.lastSpeedCheck = now - conn.lastSpeedDownload = conn.Downloaded - } else if now-conn.lastSpeedCheck >= int64(500*time.Millisecond) { - elapsed := float64(now-conn.lastSpeedCheck) / float64(time.Second) - if elapsed > 0 { - conn.speed = int64(float64(conn.Downloaded-conn.lastSpeedDownload) / elapsed) - } - conn.lastSpeedCheck = now - conn.lastSpeedDownload = conn.Downloaded - } - f.connMu.Unlock() - - if finished { - return nil } } @@ -1337,6 +1376,73 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } } +// commitResolveTakeover stops the still-live Resolve download only after the +// first Range response has been validated. Resolve may have advanced beyond +// requestedStart while the speculative request was in flight, so the overlap +// is returned for the caller to discard from the Range response body. +func (f *Fetcher) commitResolveTakeover(conn *connection, requestedStart int64) (int64, error) { + if conn.ID != 0 || !f.resolveFallback.CompareAndSwap(true, false) { + return 0, nil + } + + prefetched := f.stopPrefetchAndCopyData() + if prefetched < requestedStart || prefetched > f.meta.Res.Size { + return 0, fmt.Errorf("resolve prefetch advanced to invalid offset %d for range start %d", prefetched, requestedStart) + } + f.resolveDataPos.Store(prefetched) + + overlap := prefetched - requestedStart + f.connMu.Lock() + if f.ifRange != "" { + f.rangeValidatorPinned = true + } + conn.Chunk.Downloaded = overlap + conn.Downloaded = prefetched + f.connMu.Unlock() + return overlap, nil +} + +// completeFromResolveFallback keeps the original full response authoritative +// when the speculative first Range request cannot take over. It waits for that +// response to finish, copies it once, and converts the task to one completed +// sequential connection instead of accepting an incomplete Range result. +func (f *Fetcher) completeFromResolveFallback(conn *connection) bool { + if conn.ID != 0 || !f.resolveFallback.CompareAndSwap(true, false) { + return false + } + + for !f.prefetchDone.Load() { + if conn.ctx.Err() != nil { + return false + } + time.Sleep(10 * time.Millisecond) + } + if f.prefetchErr != nil || f.prefetchSize.Load() != f.meta.Res.Size { + f.cleanupPrefetchFile() + return false + } + + prefetched := f.stopPrefetchAndCopyData() + if prefetched != f.meta.Res.Size { + return false + } + f.resolveDataPos.Store(prefetched) + + f.connMu.Lock() + f.meta.Res.Range = false + f.rangeReprobeEligible = false + f.rangeValidatorPinned = false + conn.Chunk = newChunk(0, prefetched-1) + conn.Chunk.Downloaded = prefetched + conn.Downloaded = prefetched + conn.Completed = true + conn.State = connCompleted + conn.failed = false + conn.lastErr = nil + f.connMu.Unlock() + return true +} + func validateRangeResponse(resp *http.Response, requestedStart, requestedEnd, totalSize int64) (int64, error) { contentRange := resp.Header.Get(base.HttpHeaderContentRange) fields := strings.Fields(contentRange) @@ -2091,8 +2197,30 @@ func (f *Fetcher) Pause() error { time.Sleep(10 * time.Millisecond) } - // Clean up prefetch file - f.cleanupPrefetchFile() + // If Pause wins before the first Range response is validated, materialize + // the Resolve prefix before discarding its temporary file. The first + // connection was allocated after the earlier prefetch snapshot, so advance + // its range start to the final copied offset for a safe persisted resume. + if f.meta != nil && f.meta.Res != nil && f.meta.Res.Range && f.prefetchFile != nil { + f.resolveFallback.Store(false) + prefetched := f.stopPrefetchAndCopyData() + if prefetched > 0 { + f.resolveDataPos.Store(prefetched) + f.connMu.Lock() + if len(f.connections) == 1 { + conn := f.connections[0] + if conn != nil && conn.ID == 0 && conn.Chunk != nil && !conn.Completed { + conn.Chunk.Begin = prefetched + conn.Chunk.End = f.meta.Res.Size - 1 + conn.Chunk.Downloaded = 0 + conn.Downloaded = prefetched + } + } + f.connMu.Unlock() + } + } else { + f.cleanupPrefetchFile() + } // Clean up resolve response if still held f.resolveRespLock.Lock() diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index 8b2f66c5..dabb161b 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -1060,6 +1060,183 @@ func TestFetcher_OriginalURLFallbackPreservesResumeHeaders(t *testing.T) { resp.Body.Close() } +func TestFetcher_UsesResolveResponseWhenRangeTakeoverFails(t *testing.T) { + tests := []struct { + name string + rangeHandler func(gohttp.ResponseWriter, *gohttp.Request) + }{ + { + name: "forbidden", + rangeHandler: func(w gohttp.ResponseWriter, _ *gohttp.Request) { + w.WriteHeader(gohttp.StatusForbidden) + }, + }, + { + name: "invalid partial content", + rangeHandler: func(w gohttp.ResponseWriter, r *gohttp.Request) { + var start, end int64 + if _, err := fmt.Sscanf(r.Header.Get(base.HttpHeaderRange), "bytes=%d-%d", &start, &end); err != nil { + w.WriteHeader(gohttp.StatusBadRequest) + return + } + w.Header().Set(base.HttpHeaderContentRange, fmt.Sprintf("bytes %d-%d/%d", start+1, end, 2*1024*1024)) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", end-start)) + w.WriteHeader(gohttp.StatusPartialContent) + }, + }, + { + name: "connection closes before response", + rangeHandler: func(w gohttp.ResponseWriter, _ *gohttp.Request) { + conn, _, err := w.(gohttp.Hijacker).Hijack() + if err == nil { + conn.Close() + } + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + payload := make([]byte, 2*1024*1024) + for i := range payload { + payload[i] = byte(i % 251) + } + + var requests atomic.Int32 + var rangeRequests atomic.Int32 + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { + requestNumber := requests.Add(1) + if requestNumber == 1 { + w.Header().Set(base.HttpHeaderAcceptRanges, base.HttpHeaderBytes) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(payload))) + w.WriteHeader(gohttp.StatusOK) + w.(gohttp.Flusher).Flush() + for offset := 0; offset < len(payload); offset += 8 * 1024 { + end := min(offset+8*1024, len(payload)) + if _, err := w.Write(payload[offset:end]); err != nil { + return + } + time.Sleep(time.Millisecond) + } + return + } + + if r.Header.Get(base.HttpHeaderRange) != "" { + rangeRequests.Add(1) + } + tt.rangeHandler(w, r) + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/resolve-fallback.data"}, &base.Options{ + Path: t.TempDir(), + Name: "resolve-fallback.data", + Extra: &http.OptsExtra{ + Connections: 1, + }, + }); err != nil { + t.Fatal(err) + } + defer f.Close() + + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatalf("download should continue from the Resolve response when Range takeover fails: %v", err) + } + if rangeRequests.Load() == 0 { + t.Fatal("download did not attempt a Range request") + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("downloaded file differs from the Resolve response") + } + }) + } +} + +func TestFetcher_ValidRangeTakesOverResolveResponse(t *testing.T) { + const resolveETag = `"resolve-v1"` + payload := make([]byte, 4*1024*1024) + for i := range payload { + payload[i] = byte(i % 251) + } + + var rangeRequests atomic.Int32 + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { + rangeHeader := r.Header.Get(base.HttpHeaderRange) + w.Header().Set(base.HttpHeaderAcceptRanges, base.HttpHeaderBytes) + if rangeHeader == "" { + w.Header().Set(base.HttpHeaderETag, resolveETag) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(payload))) + w.WriteHeader(gohttp.StatusOK) + w.(gohttp.Flusher).Flush() + for offset := 0; offset < len(payload); offset += 8 * 1024 { + end := min(offset+8*1024, len(payload)) + if _, err := w.Write(payload[offset:end]); err != nil { + return + } + time.Sleep(time.Millisecond) + } + return + } + + rangeRequests.Add(1) + if got := r.Header.Get(base.HttpHeaderIfRange); got != resolveETag { + t.Errorf("If-Range = %q, want Resolve ETag %q", got, resolveETag) + } + var start, end int64 + if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end); err != nil { + w.WriteHeader(gohttp.StatusBadRequest) + return + } + w.Header().Set(base.HttpHeaderContentRange, fmt.Sprintf("bytes %d-%d/%d", start, end, len(payload))) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", end-start+1)) + w.WriteHeader(gohttp.StatusPartialContent) + for offset := start; offset <= end; offset += 8 * 1024 { + chunkEnd := min(offset+8*1024, end+1) + if _, err := w.Write(payload[offset:chunkEnd]); err != nil { + return + } + } + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/resolve-takeover.data"}, &base.Options{ + Path: t.TempDir(), + Name: "resolve-takeover.data", + Extra: &http.OptsExtra{ + Connections: 4, + }, + }); err != nil { + t.Fatal(err) + } + defer f.Close() + + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + if rangeRequests.Load() == 0 { + t.Fatal("download did not attempt a Range takeover") + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("downloaded file differs after valid Range takeover") + } +} + func TestFetcher_RejectsInvalidPartialContent(t *testing.T) { const ( requestedStart = int64(1024) @@ -1955,6 +2132,11 @@ func TestFetcher_RetryAfterError(t *testing.T) { if err != nil { t.Fatal(err) } + // This test exercises retrying a failed ranged download. Discard the live + // Resolve response first so it cannot satisfy the task as a fallback. + if err := fetcher.Pause(); err != nil { + t.Fatal(err) + } // First download attempt - should fail because server returns 416 after resolve err = fetcher.Start() @@ -2865,6 +3047,12 @@ func TestFetcher_Patch_CookieExpired(t *testing.T) { if initialExtra.Header["Cookie"] != "session=old_token" { t.Errorf("Initial Cookie = %v, want session=old_token", initialExtra.Header["Cookie"]) } + // This test exercises patching credentials after an authenticated request + // fails. Discard the already-authorized Resolve response so the stale cookie + // is observable on the first Start. + if err := f.Pause(); err != nil { + t.Fatal(err) + } // Step 2: Start download - should fail because old_token is now expired // (server only accepts old_token for first request, subsequent requests need new_token) diff --git a/pkg/download/downloader_test.go b/pkg/download/downloader_test.go index 0f98890d..9de01a78 100644 --- a/pkg/download/downloader_test.go +++ b/pkg/download/downloader_test.go @@ -167,11 +167,13 @@ func newTestDownloadOptAt(path string) *base.Options { Path: path, Name: test.DownloadName, Extra: http.OptsExtra{ - Connections: 4, + Connections: testDownloadConnections, }, } } +const testDownloadConnections = 4 + func TestDownloader_Resolve(t *testing.T) { listener := test.StartTestFileServer() defer listener.Close() @@ -1479,7 +1481,11 @@ func TestDownloader_PauseAndContinue(t *testing.T) { } func TestDownloader_PauseAllAndContinueAll(t *testing.T) { - listener := test.StartTestSlowFileServer(time.Millisecond * 2000) + const targetDownloadDuration = time.Second + // Scale the per-byte delay with the configured connection count so a full + // ranged transfer stays near one second and is still running at Pause. + delayPerByte := targetDownloadDuration * testDownloadConnections / test.BuildSize + listener := test.StartTestLowSpeedServer(delayPerByte) defer listener.Close() downloader := NewDownloader(nil)