fix(http): fall back when servers ignore range requests (#1459)

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

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) {

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