diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 1853f81a..425261b1 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -37,6 +37,7 @@ const ( var ( errRangeRequestIgnored = errors.New("server ignored HTTP range request") errInvalidRangeResponse = errors.New("invalid HTTP range response") + errIncompleteResponse = errors.New("incomplete HTTP response") ) func isRangeIntegrityError(err error) bool { @@ -88,18 +89,32 @@ const ( type chunk struct { Begin int64 - End int64 + End *int64 // nil until an open-ended Range response reveals the total size Downloaded int64 } func (c *chunk) remain() int64 { - return c.End - c.Begin + 1 - c.Downloaded + return *c.End - c.Begin + 1 - c.Downloaded +} + +func (c *chunk) openEnded() bool { + return c.End == nil +} + +func (c *chunk) setEnd(end int64) { + c.End = &end } func newChunk(begin int64, end int64) *chunk { return &chunk{ Begin: begin, - End: end, + End: &end, + } +} + +func newOpenEndedChunk(begin int64) *chunk { + return &chunk{ + Begin: begin, } } @@ -466,7 +481,7 @@ func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { // Start async prefetch in background (only for range-supported resources) // For non-range resources, the response will be used directly in Start - if res.Range && res.Size > 0 { + if res.Range { f.prefetchStopCh = make(chan struct{}) f.resolveFallback.Store(true) go f.asyncPrefetch() @@ -699,7 +714,8 @@ func (f *Fetcher) doStart() error { // For non-range resources, the response will be used directly var prefetchedBytes int64 if f.meta.Res.Range { - if f.resolveFallback.Load() && !(f.prefetchDone.Load() && f.prefetchErr == nil && f.prefetchSize.Load() >= f.meta.Res.Size) { + resolveComplete := f.meta.Res.Size > 0 && f.prefetchDone.Load() && f.prefetchErr == nil && f.prefetchSize.Load() >= f.meta.Res.Size + if f.resolveFallback.Load() && !resolveComplete { // 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. @@ -766,7 +782,7 @@ func (f *Fetcher) downloadLoop(done chan struct{}) { // Capture the mode before launching the first connection. The server may // ignore that connection's Range request and switch the fetcher to a // sequential download while the request is in flight. - startedWithRanges := f.meta.Res.Range && f.meta.Res.Size > 0 + startedWithRanges := f.meta.Res.Range // Fresh start: begin with resolve connection f.startResolveDownload() // Non-range downloads wait for their only connection in @@ -829,8 +845,10 @@ func (f *Fetcher) downloadLoop(done chan struct{}) { } func (f *Fetcher) startResolveDownload() { - // If no range support or size unknown, just use single connection with resolve response - if !f.meta.Res.Range || f.meta.Res.Size == 0 { + // Without Range support, consume the reusable Resolve response sequentially. + // A Range-capable resource with an unknown size instead starts one open-ended + // worker, which discovers the total from Content-Range before slow-start expands. + if !f.meta.Res.Range { // Create a single connection for the entire file conn := &connection{ ID: 0, @@ -875,8 +893,9 @@ func (f *Fetcher) expandConnections() { // Check if we have prefetched data prefetched := f.resolveDataPos.Load() - // If prefetched all data, mark as done - if prefetched >= totalSize { + // If prefetched all data, mark as done. A zero size means the extent is + // still unknown, so the first worker must discover it with bytes=start-. + if totalSize > 0 && prefetched >= totalSize { f.connMu.Unlock() // Close the file before signaling completion @@ -893,11 +912,15 @@ func (f *Fetcher) expandConnections() { } // First connection starts from prefetched position + connChunk := newOpenEndedChunk(prefetched) + if totalSize > 0 { + connChunk = newChunk(prefetched, totalSize-1) + } conn := &connection{ ID: 0, Role: rolePrimary, State: connNotStarted, - Chunk: newChunk(prefetched, totalSize-1), + Chunk: connChunk, } // Mark prefetched bytes as already downloaded conn.Chunk.Downloaded = 0 // Start fresh from prefetched position @@ -928,6 +951,9 @@ func (f *Fetcher) expandConnections() { if conn.Completed || conn.State == connFailed { continue } + if conn.Chunk.openEnded() { + continue + } remain := conn.Chunk.remain() // Only split if remaining work is at least 2x the minimum split size if remain > maxRemain && remain > minSplitSize*2 { @@ -942,9 +968,10 @@ func (f *Fetcher) expandConnections() { } // Split the work: new connection takes the latter half - splitPoint := maxRemainConn.Chunk.End - maxRemainConn.Chunk.remain()/2 - newChunk := newChunk(splitPoint+1, maxRemainConn.Chunk.End) - maxRemainConn.Chunk.End = splitPoint + chunkEnd := *maxRemainConn.Chunk.End + splitPoint := chunkEnd - maxRemainConn.Chunk.remain()/2 + newChunk := newChunk(splitPoint+1, chunkEnd) + maxRemainConn.Chunk.setEnd(splitPoint) connID := len(f.connections) conn := &connection{ @@ -1121,16 +1148,22 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ f.rangeValidatorPinned = false intentionalRestart = true } - if f.meta.Res.Range && conn.Chunk.remain() <= 0 { + openEndedRange := f.meta.Res.Range && conn.Chunk.openEnded() + if f.meta.Res.Range && !openEndedRange && conn.Chunk.remain() <= 0 { f.connMu.Unlock() return nil } rangeStart := conn.Chunk.Begin + conn.Chunk.Downloaded - rangeEnd := conn.Chunk.End + var rangeEnd *int64 + if !openEndedRange { + end := *conn.Chunk.End + rangeEnd = &end + } rangeMode := f.meta.Res.Range ifRange := "" if resumeProbe { - rangeEnd = f.meta.Res.Size - 1 + end := f.meta.Res.Size - 1 + rangeEnd = &end ifRange = f.ifRange } else if rangeMode && (f.rangeValidatorPinned || f.resolveFallback.Load()) { ifRange = f.ifRange @@ -1143,8 +1176,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } if rangeMode || resumeProbe { - httpReq.Header.Set(base.HttpHeaderRange, - fmt.Sprintf(base.HttpHeaderRangeFormat, rangeStart, rangeEnd)) + httpReq.Header.Set(base.HttpHeaderRange, formatRangeHeader(rangeStart, rangeEnd)) if ifRange != "" { httpReq.Header.Set(base.HttpHeaderIfRange, ifRange) } @@ -1209,11 +1241,18 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } } if rangeRequested && resp.StatusCode == base.HttpCodePartialContent { - expectedResponseLength, err = validateRangeResponse(resp, rangeStart, rangeEnd, f.meta.Res.Size) + var responseTotal int64 + expectedResponseLength, responseTotal, err = validateRangeResponse(resp, rangeStart, rangeEnd, f.meta.Res.Size) if err != nil { resp.Body.Close() return err } + if openEndedRange { + if err := f.resolveOpenEndedRange(conn, responseTotal); err != nil { + resp.Body.Close() + return err + } + } responseBytesToDiscard, err = f.commitResolveTakeover(conn, rangeStart) if err != nil { resp.Body.Close() @@ -1228,7 +1267,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ f.rangeValidatorPinned = true f.sequentialSizeUnknown = false conn.Chunk.Begin = 0 - conn.Chunk.End = f.meta.Res.Size - 1 + conn.Chunk.setEnd(f.meta.Res.Size - 1) f.connMu.Unlock() } } @@ -1367,6 +1406,9 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } } if expectedResponseLength >= 0 && responseBytesRead != expectedResponseLength { + if rangeRequested && responseBytesRead < expectedResponseLength { + return fmt.Errorf("%w: response body length %d, expected %d", errIncompleteResponse, responseBytesRead, expectedResponseLength) + } return fmt.Errorf("%w: response body length %d, expected %d", errInvalidRangeResponse, responseBytesRead, expectedResponseLength) } return nil @@ -1417,18 +1459,25 @@ func (f *Fetcher) completeFromResolveFallback(conn *connection) bool { } time.Sleep(10 * time.Millisecond) } - if f.prefetchErr != nil || f.prefetchSize.Load() != f.meta.Res.Size { + knownSize := f.meta.Res.Size > 0 + if f.prefetchErr != nil || (knownSize && f.prefetchSize.Load() != f.meta.Res.Size) { f.cleanupPrefetchFile() return false } prefetched := f.stopPrefetchAndCopyData() - if prefetched != f.meta.Res.Size { + if knownSize && prefetched != f.meta.Res.Size { return false } f.resolveDataPos.Store(prefetched) f.connMu.Lock() + if !knownSize { + f.meta.Res.Size = prefetched + if len(f.meta.Res.Files) > 0 && f.meta.Res.Files[0] != nil { + f.meta.Res.Files[0].Size = prefetched + } + } f.meta.Res.Range = false f.rangeReprobeEligible = false f.rangeValidatorPinned = false @@ -1443,40 +1492,69 @@ func (f *Fetcher) completeFromResolveFallback(conn *connection) bool { return true } -func validateRangeResponse(resp *http.Response, requestedStart, requestedEnd, totalSize int64) (int64, error) { +func validateRangeResponse(resp *http.Response, requestedStart int64, requestedEnd *int64, totalSize int64) (int64, int64, error) { contentRange := resp.Header.Get(base.HttpHeaderContentRange) fields := strings.Fields(contentRange) if len(fields) != 2 || !strings.EqualFold(fields[0], base.HttpHeaderBytes) { - return 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) + return 0, 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) } rangeAndTotal := strings.Split(fields[1], "/") if len(rangeAndTotal) != 2 { - return 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) + return 0, 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) } bounds := strings.Split(rangeAndTotal[0], "-") if len(bounds) != 2 { - return 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) + return 0, 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) } start, startErr := strconv.ParseInt(bounds[0], 10, 64) end, endErr := strconv.ParseInt(bounds[1], 10, 64) total, totalErr := strconv.ParseInt(rangeAndTotal[1], 10, 64) if startErr != nil || endErr != nil || totalErr != nil || start < 0 || end < start || total <= end { - return 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) + return 0, 0, fmt.Errorf("%w: malformed Content-Range %q", errInvalidRangeResponse, contentRange) } - if start != requestedStart || end != requestedEnd { - return 0, fmt.Errorf("%w: Content-Range %d-%d does not match requested range %d-%d", errInvalidRangeResponse, start, end, requestedStart, requestedEnd) + if start != requestedStart || (requestedEnd != nil && end != *requestedEnd) || (requestedEnd == nil && end != total-1) { + return 0, 0, fmt.Errorf("%w: Content-Range %d-%d does not match requested range", errInvalidRangeResponse, start, end) } if totalSize > 0 && total != totalSize { - return 0, fmt.Errorf("%w: Content-Range total %d does not match resource size %d", errInvalidRangeResponse, total, totalSize) + return 0, 0, fmt.Errorf("%w: Content-Range total %d does not match resource size %d", errInvalidRangeResponse, total, totalSize) } expectedLength := end - start + 1 if resp.ContentLength >= 0 && resp.ContentLength != expectedLength { - return 0, fmt.Errorf("%w: Content-Length %d does not match range length %d", errInvalidRangeResponse, resp.ContentLength, expectedLength) + return 0, 0, fmt.Errorf("%w: Content-Length %d does not match range length %d", errInvalidRangeResponse, resp.ContentLength, expectedLength) } - return expectedLength, nil + return expectedLength, total, nil +} + +func (f *Fetcher) resolveOpenEndedRange(conn *connection, totalSize int64) error { + if totalSize <= 0 { + return fmt.Errorf("%w: invalid discovered resource size %d", errInvalidRangeResponse, totalSize) + } + + // Publish the discovered extent before the response-ready signal can expand + // the slow-start batch. Store observes the same metadata/progress snapshot. + f.connMu.Lock() + f.fileMu.Lock() + if f.file == nil { + f.fileMu.Unlock() + f.connMu.Unlock() + return errors.New("target file is not open") + } + err := f.file.Truncate(totalSize) + f.fileMu.Unlock() + if err != nil { + f.connMu.Unlock() + return err + } + f.meta.Res.Size = totalSize + if len(f.meta.Res.Files) > 0 && f.meta.Res.Files[0] != nil { + f.meta.Res.Files[0].Size = totalSize + } + conn.Chunk.setEnd(totalSize - 1) + f.connMu.Unlock() + return nil } // fallbackToSequentialDownload handles origins that advertise byte ranges but @@ -1660,14 +1738,23 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) { if err != nil { if err == io.EOF { f.connMu.Lock() + if f.meta.Res.Size == 0 { + f.meta.Res.Size = conn.Downloaded + if len(f.meta.Res.Files) > 0 && f.meta.Res.Files[0] != nil { + f.meta.Res.Files[0].Size = conn.Downloaded + } + } conn.Completed = true conn.State = connCompleted f.connMu.Unlock() return } - // Reading from resolve response failed: treat as transient (do not count as fail) + // The Resolve response cannot be resumed on this path. Preserve the read + // error so an unknown-length stream is not mistaken for a clean EOF. f.connMu.Lock() conn.State = connFailed + conn.failed = true + conn.lastErr = err f.connMu.Unlock() return } @@ -1848,6 +1935,9 @@ func (f *Fetcher) helpOtherConnection(helper *connection) bool { if r == helper || r.Completed || r.State == connFailed { continue } + if r.Chunk.openEnded() { + continue + } remain := r.Chunk.remain() if remain < stealMinChunkSize { @@ -1875,10 +1965,11 @@ func (f *Fetcher) helpOtherConnection(helper *connection) bool { } // Re-calculate the chunk range: steal half of the remaining work - helper.Chunk.Begin = slowestConn.Chunk.End - slowestConn.Chunk.remain()/2 - helper.Chunk.End = slowestConn.Chunk.End + chunkEnd := *slowestConn.Chunk.End + helper.Chunk.Begin = chunkEnd - slowestConn.Chunk.remain()/2 + helper.Chunk.setEnd(chunkEnd) helper.Chunk.Downloaded = 0 - slowestConn.Chunk.End = helper.Chunk.Begin - 1 + slowestConn.Chunk.setEnd(helper.Chunk.Begin - 1) return true } @@ -1893,7 +1984,7 @@ func (f *Fetcher) resetConnectionForRestart(conn *connection) { conn.Chunk = newChunk(0, 0) } else { conn.Chunk.Begin = 0 - conn.Chunk.End = 0 + conn.Chunk.setEnd(0) conn.Chunk.Downloaded = 0 } conn.Downloaded = 0 @@ -1998,7 +2089,7 @@ func (f *Fetcher) onDownloadComplete() { for _, conn := range f.connections { needsMoreData := false if f.meta.Res.Range { - needsMoreData = conn.Chunk != nil && conn.Chunk.remain() > 0 + needsMoreData = conn.Chunk != nil && (conn.Chunk.openEnded() || conn.Chunk.remain() > 0) } else if f.meta.Res.Size > 0 { needsMoreData = conn.Downloaded < f.meta.Res.Size } else { @@ -2211,7 +2302,11 @@ func (f *Fetcher) Pause() error { 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 + if f.meta.Res.Size > 0 { + conn.Chunk.setEnd(f.meta.Res.Size - 1) + } else { + conn.Chunk.End = nil + } conn.Chunk.Downloaded = 0 conn.Downloaded = prefetched } diff --git a/internal/protocol/http/fetcher_manager.go b/internal/protocol/http/fetcher_manager.go index baf7d21e..99d9b571 100644 --- a/internal/protocol/http/fetcher_manager.go +++ b/internal/protocol/http/fetcher_manager.go @@ -138,6 +138,10 @@ func snapshotConnectionsLocked(connections []*connection) []*connection { } if conn.Chunk != nil { copyChunk := *conn.Chunk + if conn.Chunk.End != nil { + end := *conn.Chunk.End + copyChunk.End = &end + } copyConn.Chunk = ©Chunk } snapshot[i] = copyConn diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index dabb161b..c6880b14 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -1053,7 +1053,8 @@ func TestFetcher_OriginalURLFallbackPreservesResumeHeaders(t *testing.T) { config: &config{}, meta: &fetcher.FetcherMeta{Req: &base.Request{URL: server.URL}}, } - resp, err := f.tryFallbackToOriginalURL(context.Background(), server.Client(), start, end, true, ifRange) + rangeEnd := int64(end) + resp, err := f.tryFallbackToOriginalURL(context.Background(), server.Client(), start, &rangeEnd, true, ifRange) if err != nil { t.Fatal(err) } @@ -1237,6 +1238,430 @@ func TestFetcher_ValidRangeTakesOverResolveResponse(t *testing.T) { } } +func TestFetcher_UnknownSizeRangeUsesOpenEndedWorker(t *testing.T) { + const resolveETag = `"unknown-size-v1"` + payload := make([]byte, 4*1024*1024) + for i := range payload { + payload[i] = byte((i*11 + 7) % 251) + } + + var openEndedRequests atomic.Int32 + 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) + w.Header().Set(base.HttpHeaderETag, resolveETag) + if rangeHeader == "" { + 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 + } + w.(gohttp.Flusher).Flush() + time.Sleep(2 * time.Millisecond) + } + return + } + + rangeRequests.Add(1) + if got := r.Header.Get(base.HttpHeaderIfRange); got != resolveETag { + t.Errorf("If-Range = %q, want %q", got, resolveETag) + } + var start, end int64 + if strings.HasSuffix(rangeHeader, "-") { + openEndedRequests.Add(1) + if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-", &start); err != nil { + t.Errorf("parse open-ended Range %q: %v", rangeHeader, err) + w.WriteHeader(gohttp.StatusBadRequest) + return + } + end = int64(len(payload) - 1) + } else if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end); err != nil { + t.Errorf("parse bounded Range %q: %v", rangeHeader, err) + w.WriteHeader(gohttp.StatusBadRequest) + return + } + if start > end { + w.Header().Set(base.HttpHeaderContentRange, fmt.Sprintf("bytes */%d", len(payload))) + w.WriteHeader(gohttp.StatusRequestedRangeNotSatisfiable) + 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 + } + w.(gohttp.Flusher).Flush() + time.Sleep(time.Millisecond) + } + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size-range.data"}, &base.Options{ + Path: t.TempDir(), + Name: "unknown-size-range.data", + Extra: &http.OptsExtra{ + Connections: 4, + }, + }); err != nil { + t.Fatal(err) + } + defer f.Close() + + if f.meta.Res.Size != 0 || !f.meta.Res.Range { + t.Fatalf("resolved resource = %+v, want unknown size with Range support", f.meta.Res) + } + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + if got := openEndedRequests.Load(); got != 1 { + t.Fatalf("open-ended Range requests = %d, want 1", got) + } + if got := rangeRequests.Load(); got < 2 { + t.Fatalf("Range requests = %d, want slow-start expansion after size discovery", got) + } + if got, want := f.meta.Res.Size, int64(len(payload)); got != want { + t.Fatalf("resource size = %d, want %d", got, want) + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("downloaded file differs after unknown-size Range takeover") + } +} + +func TestFetcher_UnknownSizeRangeRetriesShortPartialBody(t *testing.T) { + payload := make([]byte, 1024*1024) + for i := range payload { + payload[i] = byte((i*13 + 5) % 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) + w.Header().Set(base.HttpHeaderETag, `"short-range-v1"`) + if rangeHeader == "" { + 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 + } + w.(gohttp.Flusher).Flush() + time.Sleep(2 * time.Millisecond) + } + return + } + + requestNumber := rangeRequests.Add(1) + var start, requestedEnd int64 + openEnded := strings.HasSuffix(rangeHeader, "-") + if openEnded { + if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-", &start); err != nil { + t.Errorf("parse open-ended Range %q: %v", rangeHeader, err) + w.WriteHeader(gohttp.StatusBadRequest) + return + } + requestedEnd = int64(len(payload) - 1) + } else if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &requestedEnd); err != nil { + t.Errorf("parse Range %q: %v", rangeHeader, err) + w.WriteHeader(gohttp.StatusBadRequest) + return + } + + w.Header().Set(base.HttpHeaderContentRange, fmt.Sprintf("bytes %d-%d/%d", start, requestedEnd, len(payload))) + w.WriteHeader(gohttp.StatusPartialContent) + w.(gohttp.Flusher).Flush() + if requestNumber == 1 { + // End the chunked response cleanly after only part of the declared + // Content-Range. The worker must retry from the materialized offset. + shortEnd := min(start+64*1024, requestedEnd+1) + _, _ = w.Write(payload[start:shortEnd]) + return + } + _, _ = w.Write(payload[start : requestedEnd+1]) + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/short-range.data"}, &base.Options{ + Path: t.TempDir(), + Name: "short-range.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.Fatal(err) + } + if got := rangeRequests.Load(); got < 2 { + t.Fatalf("Range requests = %d, want a retry after the short body", got) + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("downloaded file differs after retrying the short Range body") + } +} + +func TestFetcher_UnknownSizeRangeFallsBackToResolve(t *testing.T) { + payload := make([]byte, 512*1024) + for i := range payload { + payload[i] = byte((i*17 + 3) % 251) + } + + var rangeRequests atomic.Int32 + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { + w.Header().Set(base.HttpHeaderAcceptRanges, base.HttpHeaderBytes) + if r.Header.Get(base.HttpHeaderRange) == "" { + 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 + } + w.(gohttp.Flusher).Flush() + time.Sleep(time.Millisecond) + } + return + } + + rangeRequests.Add(1) + w.Header().Set(base.HttpHeaderContentRange, fmt.Sprintf("bytes 1-%d/%d", len(payload)-1, len(payload))) + w.WriteHeader(gohttp.StatusPartialContent) + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size-fallback.data"}, &base.Options{ + Path: t.TempDir(), + Name: "unknown-size-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("Wait() should use the complete Resolve response: %v", err) + } + if got := rangeRequests.Load(); got != 1 { + t.Fatalf("Range requests = %d, want 1", got) + } + if f.meta.Res.Range { + t.Fatal("Resolve fallback should switch the task to sequential mode") + } + if got, want := f.meta.Res.Size, int64(len(payload)); got != want { + t.Fatalf("resource size = %d, want %d", got, want) + } + 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 unknown-size Resolve fallback") + } +} + +func TestFetcher_UnknownSizeRangeRemainsOpenEndedAfterPause(t *testing.T) { + payload := make([]byte, 1024*1024) + for i := range payload { + payload[i] = byte((i*19 + 1) % 251) + } + + firstRangeStarted := make(chan struct{}) + 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) + w.Header().Set(base.HttpHeaderETag, `"pause-open-range-v1"`) + if rangeHeader == "" { + 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 + } + w.(gohttp.Flusher).Flush() + time.Sleep(2 * time.Millisecond) + } + return + } + + requestNumber := rangeRequests.Add(1) + if requestNumber == 1 { + close(firstRangeStarted) + <-r.Context().Done() + return + } + if !strings.HasSuffix(rangeHeader, "-") { + t.Errorf("resumed Range = %q, want an open-ended range", rangeHeader) + w.WriteHeader(gohttp.StatusBadRequest) + return + } + var start int64 + if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-", &start); err != nil { + t.Errorf("parse resumed Range %q: %v", rangeHeader, err) + w.WriteHeader(gohttp.StatusBadRequest) + return + } + end := int64(len(payload) - 1) + 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) + _, _ = w.Write(payload[start:]) + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/pause-open-range.data"}, &base.Options{ + Path: t.TempDir(), + Name: "pause-open-range.data", + Extra: &http.OptsExtra{ + Connections: 1, + }, + }); err != nil { + t.Fatal(err) + } + defer f.Close() + + if err := f.Start(); err != nil { + t.Fatal(err) + } + select { + case <-firstRangeStarted: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the first open-ended worker") + } + if err := f.Pause(); err != nil { + t.Fatal(err) + } + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + if got := rangeRequests.Load(); got != 2 { + t.Fatalf("Range requests = %d, want an open-ended retry after pause", got) + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("downloaded file differs after pausing an unknown-size Range") + } +} + +func TestFetcher_UnknownSizeResolveReadErrorFailsTask(t *testing.T) { + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, _ *gohttp.Request) { + conn, rw, err := w.(gohttp.Hijacker).Hijack() + if err != nil { + t.Errorf("hijack response: %v", err) + return + } + defer conn.Close() + _, _ = fmt.Fprint(rw, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n800\r\n") + _, _ = rw.Write(make([]byte, 1024)) + _ = rw.Flush() + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/broken-chunked.data"}, &base.Options{ + Path: t.TempDir(), + Name: "broken-chunked.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.Fatal("Wait() succeeded after the unknown-size response ended unexpectedly") + } + if got := f.getState(); got != stateError { + t.Fatalf("fetcher state = %v, want %v", got, stateError) + } +} + +func TestFetcher_UnknownSizeResolveCleanEOFUpdatesSize(t *testing.T) { + payload := bytes.Repeat([]byte("gopeed-unknown-size\n"), 8192) + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, _ *gohttp.Request) { + w.WriteHeader(gohttp.StatusOK) + w.(gohttp.Flusher).Flush() + _, _ = w.Write(payload) + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size.data"}, &base.Options{ + Path: t.TempDir(), + Name: "unknown-size.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.Fatal(err) + } + if got, want := f.meta.Res.Size, int64(len(payload)); got != want { + t.Fatalf("resource size = %d, want %d", got, want) + } + if got, want := f.meta.Res.Files[0].Size, int64(len(payload)); got != want { + t.Fatalf("file size = %d, want %d", got, want) + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("downloaded file differs for a clean unknown-size response") + } +} + func TestFetcher_RejectsInvalidPartialContent(t *testing.T) { const ( requestedStart = int64(1024) @@ -1314,14 +1739,6 @@ func TestFetcher_RejectsInvalidPartialContent(t *testing.T) { bodyLength: int(rangeLength - 1), wantError: true, }, - { - name: "short chunked body", - contentRange: "bytes 1024-2047/4096", - contentLength: -1, - bodyLength: int(rangeLength - 1), - chunked: true, - wantError: true, - }, { name: "long chunked body", contentRange: "bytes 1024-2047/4096", @@ -2342,7 +2759,7 @@ func TestFetcherManager_RestoreUsesSnapshotRangeMode(t *testing.T) { ID: 0, Role: rolePrimary, State: connFailed, - Chunk: &chunk{Begin: 0, End: 99, Downloaded: 64}, + Chunk: newChunk(0, 99), Downloaded: 64, }}, IfRange: `"snapshot-v1"`, @@ -2352,6 +2769,7 @@ func TestFetcherManager_RestoreUsesSnapshotRangeMode(t *testing.T) { ResourceSize: &resourceSize, FileSize: &fileSize, } + saved.Connections[0].Chunk.Downloaded = 64 encoded, err := json.Marshal(saved) if err != nil { t.Fatal(err) @@ -2457,7 +2875,7 @@ func TestFetcherManager_RestoreUsesAtomicSizeSnapshotForDownload(t *testing.T) { ID: 0, Role: rolePrimary, State: connFailed, - Chunk: &chunk{Begin: 0, End: resourceSize - 1, Downloaded: savedPrefix}, + Chunk: newChunk(0, resourceSize-1), Downloaded: savedPrefix, }}, IfRange: `"old-v1"`, @@ -2466,6 +2884,7 @@ func TestFetcherManager_RestoreUsesAtomicSizeSnapshotForDownload(t *testing.T) { ResourceSize: &resourceSize, FileSize: &fileSize, } + saved.Connections[0].Chunk.Downloaded = savedPrefix // Simulate task metadata serialized after an interrupted chunked replacement // while the fetcher snapshot still describes the preceding representation. staleMeta := &fetcher.FetcherMeta{ diff --git a/internal/protocol/http/helper.go b/internal/protocol/http/helper.go index 1b796a62..1292723e 100644 --- a/internal/protocol/http/helper.go +++ b/internal/protocol/http/helper.go @@ -205,15 +205,14 @@ func isRedirectExpiredError(err error) bool { // tryFallbackToOriginalURL attempts to make a request using the original URL // when the redirect URL has expired. Returns the response if successful. -func (f *Fetcher) tryFallbackToOriginalURL(ctx context.Context, client *http.Client, rangeStart, rangeEnd int64, rangeRequested bool, ifRange string) (*http.Response, error) { +func (f *Fetcher) tryFallbackToOriginalURL(ctx context.Context, client *http.Client, rangeStart int64, rangeEnd *int64, rangeRequested bool, ifRange string) (*http.Response, error) { httpReq, err := f.buildRequestWithOriginalURL(ctx, f.meta.Req) if err != nil { return nil, err } if rangeRequested { - httpReq.Header.Set(base.HttpHeaderRange, - fmt.Sprintf(base.HttpHeaderRangeFormat, rangeStart, rangeEnd)) + httpReq.Header.Set(base.HttpHeaderRange, formatRangeHeader(rangeStart, rangeEnd)) if ifRange != "" { httpReq.Header.Set(base.HttpHeaderIfRange, ifRange) } @@ -232,6 +231,13 @@ func (f *Fetcher) tryFallbackToOriginalURL(ctx context.Context, client *http.Cli return resp, nil } +func formatRangeHeader(start int64, end *int64) string { + if end == nil { + return fmt.Sprintf("bytes=%d-", start) + } + return fmt.Sprintf(base.HttpHeaderRangeFormat, start, *end) +} + // buildClient creates an HTTP client with the default connection timeout. // Used for resolve phase where we don't have connection time data yet. func (f *Fetcher) buildClient() *http.Client {