diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index e5604a04..f504e725 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -34,6 +34,15 @@ const ( stealMinChunkSize = 512 * 1024 // Min steal size: 512KB (avoid tiny chunks) ) +var ( + errRangeRequestIgnored = errors.New("server ignored HTTP range request") + errInvalidRangeResponse = errors.New("invalid HTTP range response") +) + +func isRangeIntegrityError(err error) bool { + return errors.Is(err, errRangeRequestIgnored) || errors.Is(err, errInvalidRangeResponse) +} + // ============================================================================ // State Machine // ============================================================================ @@ -717,12 +726,16 @@ func (f *Fetcher) downloadLoop() { isResume := len(f.connections) > 0 if !isResume { + // 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 // Fresh start: begin with resolve connection f.startResolveDownload() // Non-range downloads wait for their only connection in // startResolveDownload. Falling through would consume the connection's // expansion signal and publish the same completion a second time. - if !f.meta.Res.Range || f.meta.Res.Size == 0 || f.getState() == stateDone { + if !startedWithRanges || f.getState() == stateDone { return } } else { @@ -924,6 +937,13 @@ func (f *Fetcher) runConnection(conn *connection) { // Rebuild client with updated fast-fail timeout on retries if retries > 0 { client = f.buildFastFailClient() + + // A sequential response always starts at byte zero. If a read failed + // after falling back from Range mode, overwrite from the beginning on + // the next request instead of appending duplicate prefix bytes. + f.connMu.Lock() + f.resetConnectionForRestart(conn) + f.connMu.Unlock() } err := f.downloadChunkOnce(conn, client, buf) @@ -945,6 +965,17 @@ func (f *Fetcher) runConnection(conn *connection) { if errors.Is(err, context.Canceled) { return } + if isRangeIntegrityError(err) { + f.connMu.Lock() + conn.lastErr = err + conn.State = connFailed + conn.failed = true + f.connMu.Unlock() + if f.slowStart != nil { + f.slowStart.onConnectFailed() + } + return + } if re := extractRequestError(err); re != nil { conn.lastErr = re @@ -1018,6 +1049,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, rangeStart, rangeEnd)) } + rangeRequested := httpReq.Header.Get(base.HttpHeaderRange) != "" // Record connection start time for adaptive timeout tracking connStartTime := time.Now() @@ -1053,6 +1085,28 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ return originalErr } } + + expectedResponseLength := int64(-1) + if rangeRequested && resp.StatusCode == base.HttpCodeOK { + if err := f.fallbackToSequentialDownload(conn); err != nil { + resp.Body.Close() + return fmt.Errorf("%w: expected 206 Partial Content, got 200 OK", err) + } + } + if rangeRequested && resp.StatusCode == base.HttpCodePartialContent { + expectedResponseLength, err = validateRangeResponse(resp, rangeStart, rangeEnd, f.meta.Res.Size) + if err != nil { + resp.Body.Close() + return err + } + } + if !f.meta.Res.Range && resp.StatusCode == base.HttpCodeOK && f.meta.Res.Size > 0 { + expectedResponseLength = f.meta.Res.Size + if resp.ContentLength >= 0 && resp.ContentLength != expectedResponseLength { + resp.Body.Close() + return fmt.Errorf("%w: Content-Length %d does not match resource size %d", errInvalidRangeResponse, resp.ContentLength, expectedResponseLength) + } + } defer resp.Body.Close() // Record successful connection time for adaptive timeout @@ -1073,6 +1127,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } reader := NewTimeoutReader(resp.Body, readTimeout) + var responseBytesRead int64 for { if conn.ctx.Err() != nil { return conn.ctx.Err() @@ -1080,6 +1135,11 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ n, err := reader.Read(buf) if n > 0 { + responseBytesRead += int64(n) + if expectedResponseLength >= 0 && responseBytesRead > expectedResponseLength { + return fmt.Errorf("%w: response body exceeds expected length %d", errInvalidRangeResponse, expectedResponseLength) + } + finished := false var writeOffset int64 @@ -1138,6 +1198,9 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ if err != nil { if err == io.EOF { + if expectedResponseLength >= 0 && responseBytesRead != expectedResponseLength { + return fmt.Errorf("%w: response body length %d, expected %d", errInvalidRangeResponse, responseBytesRead, expectedResponseLength) + } return nil } return err @@ -1145,6 +1208,65 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ } } +func validateRangeResponse(resp *http.Response, requestedStart, requestedEnd, totalSize 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) + } + + rangeAndTotal := strings.Split(fields[1], "/") + if len(rangeAndTotal) != 2 { + return 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) + } + + 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) + } + 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 totalSize > 0 && total != totalSize { + return 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 expectedLength, nil +} + +// fallbackToSequentialDownload handles origins that advertise byte ranges but +// ignore the first Range request and return the full representation with 200. +// Slow start launches only the primary connection initially, so it is safe to +// reset the prefetch offset and consume that response from byte zero. If the +// server changes behavior after parallel connections have started, fail rather +// than risk assembling a corrupt file. +func (f *Fetcher) fallbackToSequentialDownload(conn *connection) error { + f.connMu.Lock() + defer f.connMu.Unlock() + + if !f.meta.Res.Range { + return nil + } + if conn.ID != 0 || len(f.connections) != 1 { + return errRangeRequestIgnored + } + + f.meta.Res.Range = false + f.resetConnectionForRestart(conn) + f.resolveDataPos.Store(0) + return nil +} + // runConnectionWithResolveResp uses the response body from Resolve phase // This is crucial for one-time URLs that can only be accessed once func (f *Fetcher) runConnectionWithResolveResp(conn *connection) { @@ -1553,9 +1675,18 @@ func (f *Fetcher) onDownloadComplete() { // If total downloaded matches file size, consider it a success regardless of connection failures downloadComplete := f.meta.Res.Size > 0 && totalDownloaded >= f.meta.Res.Size - // Check for any errors, but ignore 403 (server connection limit) errors if download completed + // Integrity errors must win over byte-count completion. A malformed response + // may be detected only after the expected bytes have already been written. var finalErr error - if !downloadComplete && !allChunksComplete { + for _, conn := range f.connections { + if conn.State == connFailed && conn.failed && isRangeIntegrityError(conn.lastErr) { + finalErr = fmt.Errorf("connection %d failed: retries=%d, err=%w", conn.ID, conn.retryTimes, conn.lastErr) + break + } + } + + // Check for other errors, but ignore 403 (server connection limit) errors if download completed. + if finalErr == nil && !downloadComplete && !allChunksComplete { for _, conn := range f.connections { if conn.State == connFailed && conn.failed { // Skip 403 errors (server connection limit) - these are expected when exceeding server's limit diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index b5ad1659..be84a8fa 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -1,7 +1,9 @@ package http import ( + "context" "encoding/json" + "errors" "fmt" "net" gohttp "net/http" @@ -441,6 +443,317 @@ func TestFetcher_DownloadNoRangeLeavesNoStaleCompletion(t *testing.T) { } } +func TestFetcher_DownloadIgnoredRangeFallsBackToSequential(t *testing.T) { + for _, connections := range []int{1, 4} { + t.Run(fmt.Sprintf("connections_%d", connections), func(t *testing.T) { + os.Remove(test.DownloadFile) + t.Cleanup(func() { os.Remove(test.DownloadFile) }) + + listener := test.StartTestIgnoredRangeServer(time.Millisecond) + defer listener.Close() + + f := downloadReady(listener, connections, t).(*Fetcher) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + + if f.meta.Res.Range { + t.Fatal("expected ignored Range response to disable range mode") + } + stats := f.Stats().(*http.Stats) + if len(stats.Connections) != 1 { + t.Fatalf("expected one sequential connection, got %d", len(stats.Connections)) + } + if got, want := test.FileMd5(test.DownloadFile), test.FileMd5(test.BuildFile); got != want { + t.Fatalf("download checksum = %s, want %s", got, want) + } + }) + } +} + +func TestFetcher_RejectsInvalidPartialContent(t *testing.T) { + const ( + requestedStart = int64(1024) + requestedEnd = int64(2047) + totalSize = int64(4096) + rangeLength = requestedEnd - requestedStart + 1 + ) + + tests := []struct { + name string + contentRange string + contentLength int64 + bodyLength int + chunked bool + wantError bool + }{ + { + name: "valid", + contentRange: "bytes 1024-2047/4096", + contentLength: rangeLength, + bodyLength: int(rangeLength), + }, + { + name: "missing content range", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "invalid unit", + contentRange: "items 1024-2047/4096", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "malformed content range", + contentRange: "bytes 1024/4096", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "missing total separator", + contentRange: "bytes 1024-2047", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "unknown total size", + contentRange: "bytes 1024-2047/*", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "shifted content range", + contentRange: "bytes 0-1023/4096", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "wrong total size", + contentRange: "bytes 1024-2047/8192", + contentLength: rangeLength, + bodyLength: int(rangeLength), + wantError: true, + }, + { + name: "content length mismatch", + contentRange: "bytes 1024-2047/4096", + contentLength: rangeLength - 1, + 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", + contentLength: -1, + bodyLength: int(rangeLength + 1), + chunked: true, + wantError: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + server := httptest.NewServer(gohttp.HandlerFunc(func(writer gohttp.ResponseWriter, request *gohttp.Request) { + if got, want := request.Header.Get(base.HttpHeaderRange), "bytes=1024-2047"; got != want { + t.Errorf("Range header = %q, want %q", got, want) + } + if tc.contentRange != "" { + writer.Header().Set(base.HttpHeaderContentRange, tc.contentRange) + } + if tc.contentLength >= 0 { + writer.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", tc.contentLength)) + } + writer.WriteHeader(base.HttpCodePartialContent) + if tc.chunked { + writer.(gohttp.Flusher).Flush() + } + _, _ = writer.Write([]byte(strings.Repeat("x", tc.bodyLength))) + })) + defer server.Close() + + f := buildFetcher() + f.meta.Req = &base.Request{URL: server.URL} + f.meta.Res = &base.Resource{Range: true, Size: totalSize} + output, err := os.CreateTemp(t.TempDir(), "range-response-*") + if err != nil { + t.Fatal(err) + } + defer output.Close() + f.file = output + + conn := &connection{ + ID: 0, + Role: rolePrimary, + State: connNotStarted, + Chunk: newChunk(requestedStart, requestedEnd), + ctx: context.Background(), + } + f.connections = []*connection{conn} + f.wg.Add(1) + go f.runConnection(conn) + f.wg.Wait() + + if tc.wantError { + if !conn.failed || conn.State != connFailed || conn.Completed { + t.Fatalf("connection state = %v, failed = %v, completed = %v; want permanent failure", conn.State, conn.failed, conn.Completed) + } + if !errors.Is(conn.lastErr, errInvalidRangeResponse) { + t.Fatalf("connection error = %v, want %v", conn.lastErr, errInvalidRangeResponse) + } + return + } + + if conn.failed || conn.State != connCompleted || !conn.Completed { + t.Fatalf("connection state = %v, failed = %v, completed = %v; want success", conn.State, conn.failed, conn.Completed) + } + if conn.Downloaded != rangeLength { + t.Fatalf("downloaded = %d, want %d", conn.Downloaded, rangeLength) + } + }) + } +} + +func TestFetcher_InvalidResponseLengthFailsTask(t *testing.T) { + const ( + totalSize = int64(4096) + requestedStart = int64(1024) + requestedEnd = totalSize - 1 + rangeLength = requestedEnd - requestedStart + 1 + ) + + tests := []struct { + name string + status int + contentRange string + contentLength int64 + bodyChunks []int + resolvedDownloaded int64 + wantDownloaded int64 + }{ + { + name: "sequential declared short body", + status: base.HttpCodeOK, + contentLength: 1024, + bodyChunks: []int{1024}, + wantDownloaded: 0, + }, + { + name: "sequential short chunked body", + status: base.HttpCodeOK, + contentLength: -1, + bodyChunks: []int{1024}, + wantDownloaded: 1024, + }, + { + name: "sequential long chunked body", + status: base.HttpCodeOK, + contentLength: -1, + bodyChunks: []int{int(totalSize), 1}, + wantDownloaded: totalSize, + }, + { + name: "partial content long chunked body", + status: base.HttpCodePartialContent, + contentRange: "bytes 1024-4095/4096", + contentLength: -1, + bodyChunks: []int{int(rangeLength), 1}, + resolvedDownloaded: requestedStart, + wantDownloaded: rangeLength, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + server := httptest.NewServer(gohttp.HandlerFunc(func(writer gohttp.ResponseWriter, request *gohttp.Request) { + if got, want := request.Header.Get(base.HttpHeaderRange), "bytes=1024-4095"; got != want { + t.Errorf("Range header = %q, want %q", got, want) + } + if tc.contentRange != "" { + writer.Header().Set(base.HttpHeaderContentRange, tc.contentRange) + } + if tc.contentLength >= 0 { + writer.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", tc.contentLength)) + } + writer.WriteHeader(tc.status) + writer.(gohttp.Flusher).Flush() + for i, chunkSize := range tc.bodyChunks { + _, _ = writer.Write([]byte(strings.Repeat("x", chunkSize))) + writer.(gohttp.Flusher).Flush() + if i+1 < len(tc.bodyChunks) { + time.Sleep(50 * time.Millisecond) + } + } + })) + defer server.Close() + + f := buildFetcher() + f.meta.Req = &base.Request{URL: server.URL} + f.meta.Res = &base.Resource{Range: true, Size: totalSize} + output, err := os.CreateTemp(t.TempDir(), "range-task-*") + if err != nil { + t.Fatal(err) + } + if err := output.Truncate(totalSize); err != nil { + t.Fatal(err) + } + f.file = output + + conn := &connection{ + ID: 0, + Role: rolePrimary, + State: connNotStarted, + Chunk: newChunk(requestedStart, requestedEnd), + ctx: context.Background(), + } + f.connections = []*connection{conn} + if tc.resolvedDownloaded > 0 { + f.resolveConn = &connection{ + Role: roleResolve, + State: connCompleted, + Downloaded: tc.resolvedDownloaded, + Completed: true, + } + } + + f.wg.Add(1) + go f.runConnection(conn) + f.wg.Wait() + f.onDownloadComplete() + + if err := f.Wait(); !errors.Is(err, errInvalidRangeResponse) { + t.Fatalf("Wait() error = %v, want %v", err, errInvalidRangeResponse) + } + if f.getState() != stateError { + t.Fatalf("fetcher state = %v, want %v", f.getState(), stateError) + } + if conn.Completed || conn.State != connFailed || !conn.failed { + t.Fatalf("connection state = %v, failed = %v, completed = %v; want permanent failure", conn.State, conn.failed, conn.Completed) + } + if conn.Downloaded != tc.wantDownloaded { + t.Fatalf("downloaded = %d, want %d", conn.Downloaded, tc.wantDownloaded) + } + }) + } +} + func TestFetcher_DownloadChunked(t *testing.T) { listener := test.StartTestCustomServer() defer listener.Close() @@ -555,7 +868,14 @@ func TestFetcher_DownloadOnBugFileServer(t *testing.T) { defer listener.Close() downloadNormal(listener, 1, t) - downloadNormal(listener, 4, t) + + fetcher := downloadReady(listener, 4, t) + if err := fetcher.Start(); err != nil { + t.Fatal(err) + } + if err := fetcher.Wait(); err == nil || !strings.Contains(err.Error(), errInvalidRangeResponse.Error()) { + t.Fatalf("Wait() error = %v, want %v", err, errInvalidRangeResponse) + } } func TestFetcher_DownloadResume(t *testing.T) { diff --git a/internal/test/httptest.go b/internal/test/httptest.go index 2b476499..b2b496d6 100644 --- a/internal/test/httptest.go +++ b/internal/test/httptest.go @@ -372,6 +372,45 @@ func StartTestNoRangeSlowServer(delayPerChunk time.Duration) net.Listener { }) } +// StartTestIgnoredRangeServer advertises byte-range support but ignores Range +// requests and always returns the full representation with 200 OK. +func StartTestIgnoredRangeServer(delayPerChunk time.Duration) net.Listener { + return startTestServer(func(sl *shutdownListener) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Length", fmt.Sprintf("%d", BuildSize)) + writer.Header().Set("Accept-Ranges", "bytes") + writer.WriteHeader(http.StatusOK) + if flusher, ok := writer.(http.Flusher); ok { + flusher.Flush() + } + + file, err := os.Open(BuildFile) + if err != nil { + return + } + defer file.Close() + + buf := make([]byte, 256*1024) + for !sl.isShutdown { + n, readErr := file.Read(buf) + if n > 0 { + if _, writeErr := writer.Write(buf[:n]); writeErr != nil { + return + } + if delayPerChunk > 0 { + time.Sleep(delayPerChunk) + } + } + if readErr != nil { + return + } + } + }) + return mux + }) +} + // StartTestExpiringRedirectServer creates a server that simulates expiring redirect URLs. // The original URL redirects to a temporary URL that expires after a specified number of requests. // When the temporary URL expires (returns 403), the client should retry with the original URL