From f49e2510f51d1cdd605988b5d805b14155a2a259 Mon Sep 17 00:00:00 2001 From: ggg <90907031+Pseudoataraxia@users.noreply.github.com> Date: Wed, 2 Sep 2026 14:29:48 +0800 Subject: [PATCH] fix(http): re-probe range support after sequential retry (#1464) --- internal/protocol/http/fetcher.go | 299 +++++++- internal/protocol/http/fetcher_manager.go | 88 ++- internal/protocol/http/fetcher_test.go | 851 +++++++++++++++++++++- internal/protocol/http/helper.go | 46 +- pkg/base/constants.go | 2 + 5 files changed, 1245 insertions(+), 41 deletions(-) diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 375f4cb9..b626db0d 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -43,6 +43,10 @@ func isRangeIntegrityError(err error) bool { return errors.Is(err, errRangeRequestIgnored) || errors.Is(err, errInvalidRangeResponse) } +func isTerminalRangeError(err error) bool { + return isRangeIntegrityError(err) +} + // ============================================================================ // State Machine // ============================================================================ @@ -108,6 +112,7 @@ type connection struct { Completed bool failed bool + exited bool retryTimes int lastErr error @@ -214,6 +219,13 @@ func (s *slowStartController) commitBatch(count int) { s.batchReady = 0 } +func (s *slowStartController) signalExpansion() { + select { + case s.expansionCh <- struct{}{}: + default: + } +} + // ============================================================================ // Fetcher // ============================================================================ @@ -264,10 +276,14 @@ type Fetcher struct { prefetchStopCh chan struct{} // Signal to stop prefetch // Target file - file *os.File - fileMu sync.Mutex - redirectURL string - redirectLock sync.Mutex + file *os.File + fileMu sync.Mutex + redirectURL string + redirectLock sync.Mutex + ifRange string + rangeReprobeEligible bool + rangeValidatorPinned bool + sequentialSizeUnknown bool // Lifecycle control ctx context.Context @@ -401,9 +417,11 @@ func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { var lastModifiedTime *time.Time lastModified := resp.Header.Get(base.HttpHeaderLastModified) if lastModified != "" { - t, _ := time.Parse(time.RFC1123, lastModified) - lastModifiedTime = &t + if t, err := http.ParseTime(lastModified); err == nil { + lastModifiedTime = &t + } } + f.ifRange = extractIfRangeValidator(resp.Header) file := &base.FileInfo{ Size: res.Size, @@ -641,7 +659,9 @@ func (f *Fetcher) doStart() error { for _, conn := range f.connections { // Reset connections that can be retried if !conn.Completed && conn.State != connCompleted { - f.resetConnectionForRestart(conn) + if !f.hasSequentialPrefixLocked(conn) { + f.resetConnectionForRestart(conn) + } conn.State = connNotStarted conn.failed = false conn.retryTimes = 0 @@ -743,10 +763,18 @@ func (f *Fetcher) downloadLoop() { return } } else { - // Resume: restart existing connections + // A retained sequential prefix starts as the first slow-start batch. If + // its guarded request restores Range support, the normal expansion loop + // can split the remaining suffix across the configured connections. + canRecoverRange := f.hasSequentialRecoveryCandidate() + if canRecoverRange { + f.slowStart.commitBatch(1) + } f.resumeConnections() - f.waitForCompletion(ctx) - return + if !canRecoverRange { + f.waitForCompletion(ctx) + return + } } // Slow start loop @@ -756,6 +784,19 @@ func (f *Fetcher) downloadLoop() { // Paused or cancelled return case <-f.slowStart.expansionCh: + f.connMu.Lock() + rangeMode := f.meta.Res.Range + sequentialExited := len(f.connections) == 1 && f.connections[0].exited + f.connMu.Unlock() + if !rangeMode { + // The primary response is still being consumed sequentially. A + // successful re-probe or final connection exit will signal again. + if sequentialExited { + f.waitForCompletion(ctx) + return + } + continue + } // Batch completed, try to expand if f.checkCompletion() { // All work is done, wait for connections to finish @@ -924,9 +965,22 @@ func (f *Fetcher) expandConnections() { } func (f *Fetcher) runConnection(conn *connection) { - defer f.wg.Done() + defer func() { + // A range-advertising origin can fall back to one sequential response. + // Wake the download loop after that connection exits because its earlier + // 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 + f.connMu.Unlock() + f.wg.Done() + if sequentialRecovery && f.slowStart != nil { + f.slowStart.signalExpansion() + } + }() f.connMu.Lock() + conn.exited = false conn.State = connConnecting f.connMu.Unlock() @@ -942,11 +996,13 @@ func (f *Fetcher) runConnection(conn *connection) { 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. + // Preserve an eligible sequential prefix long enough to probe whether + // the origin has recovered byte-range support. downloadChunkOnce resets + // it only when no safe validator exists or If-Range returns a full 200. f.connMu.Lock() - f.resetConnectionForRestart(conn) + if !f.hasSequentialPrefixLocked(conn) { + f.resetConnectionForRestart(conn) + } f.connMu.Unlock() } @@ -969,7 +1025,7 @@ func (f *Fetcher) runConnection(conn *connection) { if errors.Is(err, context.Canceled) { return } - if isRangeIntegrityError(err) { + if isTerminalRangeError(err) { f.connMu.Lock() conn.lastErr = err conn.State = connFailed @@ -1033,15 +1089,34 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ return conn.ctx.Err() } - // Read chunk boundaries under lock to get a consistent snapshot - // This protects against concurrent modification by helpOtherConnection + // Read chunk boundaries under lock to get a consistent snapshot. A prefix + // without a safe validator cannot be resumed, so restart it before building + // the request; the returned full response will overwrite from byte zero. f.connMu.Lock() + sequentialSizeUnknown := f.sequentialSizeUnknown + hasSequentialPrefix := f.hasSequentialPrefixLocked(conn) + resumeProbe := f.canProbeSequentialResumeLocked(conn) + intentionalRestart := sequentialSizeUnknown + if sequentialSizeUnknown || (hasSequentialPrefix && !resumeProbe) { + f.resetConnectionForRestart(conn) + f.resolveDataPos.Store(0) + f.rangeValidatorPinned = false + intentionalRestart = true + } if f.meta.Res.Range && conn.Chunk.remain() <= 0 { f.connMu.Unlock() return nil } rangeStart := conn.Chunk.Begin + conn.Chunk.Downloaded rangeEnd := conn.Chunk.End + rangeMode := f.meta.Res.Range + ifRange := "" + if resumeProbe { + rangeEnd = f.meta.Res.Size - 1 + ifRange = f.ifRange + } else if rangeMode && f.rangeValidatorPinned { + ifRange = f.ifRange + } f.connMu.Unlock() httpReq, err := f.buildRequest(conn.ctx, f.meta.Req) @@ -1049,9 +1124,12 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ return err } - if f.meta.Res.Range { + if rangeMode || resumeProbe { httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, rangeStart, rangeEnd)) + if ifRange != "" { + httpReq.Header.Set(base.HttpHeaderIfRange, ifRange) + } } rangeRequested := httpReq.Header.Get(base.HttpHeaderRange) != "" @@ -1070,7 +1148,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ // Check if this might be a redirect URL expiration error // If so, try falling back to the original URL if f.hasRedirectURL() && isRedirectExpiredError(originalErr) { - fallbackResp, fallbackErr := f.tryFallbackToOriginalURL(conn.ctx, client, rangeStart, rangeEnd) + fallbackResp, fallbackErr := f.tryFallbackToOriginalURL(conn.ctx, client, rangeStart, rangeEnd, rangeRequested, ifRange) if fallbackErr == nil && fallbackResp != nil { // Fallback succeeded, use this response instead resp = fallbackResp @@ -1092,9 +1170,17 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ 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 resumeProbe { + // If-Range returning 200 provides the complete current representation. + // Discard the old prefix counters and consume this response from byte 0. + f.restartSequentialDownload(conn, extractIfRangeValidator(resp.Header)) + intentionalRestart = true + } else { + if err := f.fallbackToSequentialDownload(conn, extractIfRangeValidator(resp.Header)); err != nil { + resp.Body.Close() + return fmt.Errorf("%w: expected 206 Partial Content, got 200 OK", err) + } + intentionalRestart = true } } if rangeRequested && resp.StatusCode == base.HttpCodePartialContent { @@ -1103,12 +1189,46 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ 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. + f.connMu.Lock() + f.meta.Res.Range = true + f.rangeReprobeEligible = false + f.rangeValidatorPinned = true + f.sequentialSizeUnknown = false + conn.Chunk.Begin = 0 + conn.Chunk.End = f.meta.Res.Size - 1 + f.connMu.Unlock() + } } - 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) + if !rangeRequested && resp.StatusCode == base.HttpCodeOK { + f.connMu.Lock() + if f.rangeReprobeEligible { + // This full response might itself become the retained prefix if its + // body is interrupted, so bind any later If-Range probe to it. + f.ifRange = extractIfRangeValidator(resp.Header) + } + f.connMu.Unlock() + } + if !f.meta.Res.Range && resp.StatusCode == base.HttpCodeOK { + if intentionalRestart { + if resp.ContentLength >= 0 { + if err := f.resizeSequentialResource(resp.ContentLength); err != nil { + resp.Body.Close() + return err + } + expectedResponseLength = resp.ContentLength + } else if err := f.beginUnknownSizeSequentialResource(); err != nil { + resp.Body.Close() + return err + } + } else if 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() @@ -1202,6 +1322,11 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ if err != nil { if err == io.EOF { + if intentionalRestart && resp.ContentLength < 0 { + if err := f.resizeSequentialResource(responseBytesRead); err != nil { + return err + } + } if expectedResponseLength >= 0 && responseBytesRead != expectedResponseLength { return fmt.Errorf("%w: response body length %d, expected %d", errInvalidRangeResponse, responseBytesRead, expectedResponseLength) } @@ -1254,7 +1379,7 @@ func validateRangeResponse(resp *http.Response, requestedStart, requestedEnd, 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 { +func (f *Fetcher) fallbackToSequentialDownload(conn *connection, ifRange string) error { f.connMu.Lock() defer f.connMu.Unlock() @@ -1266,11 +1391,93 @@ func (f *Fetcher) fallbackToSequentialDownload(conn *connection) error { } f.meta.Res.Range = false + f.rangeReprobeEligible = true + f.rangeValidatorPinned = false + f.ifRange = ifRange f.resetConnectionForRestart(conn) f.resolveDataPos.Store(0) return nil } +// restartSequentialDownload consumes an If-Range 200 response as the complete +// current representation. The old prefix is no longer trusted, but this body +// can safely overwrite it from byte zero. +func (f *Fetcher) restartSequentialDownload(conn *connection, ifRange string) { + f.connMu.Lock() + defer f.connMu.Unlock() + + f.meta.Res.Range = false + f.rangeReprobeEligible = true + f.rangeValidatorPinned = false + f.ifRange = ifRange + f.resetConnectionForRestart(conn) + f.resolveDataPos.Store(0) +} + +// resizeSequentialResource makes a full restart response authoritative for +// both progress accounting and the target file. Truncation removes stale tail +// bytes when the new representation is shorter and preallocates when longer. +func (f *Fetcher) resizeSequentialResource(size int64) error { + if size < 0 { + return fmt.Errorf("invalid sequential response size %d", size) + } + + // Keep the file mutation and its recovery metadata in one connMu critical + // section so Store cannot persist one side of the transition without the + // other. Other download paths never hold fileMu while acquiring connMu. + 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(size) + f.fileMu.Unlock() + if err != nil { + f.connMu.Unlock() + return err + } + + f.meta.Res.Size = size + if len(f.meta.Res.Files) > 0 { + f.meta.Res.Files[0].Size = size + } + f.sequentialSizeUnknown = false + f.connMu.Unlock() + return nil +} + +// beginUnknownSizeSequentialResource discards all stale bytes and records that +// a restarted chunked response has no authoritative total size yet. If this +// response is interrupted, the persisted flag forces the next attempt to start +// from byte zero instead of validating a Range response against an old size. +func (f *Fetcher) beginUnknownSizeSequentialResource() error { + 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(0) + f.fileMu.Unlock() + if err != nil { + f.connMu.Unlock() + return err + } + + f.meta.Res.Size = 0 + if len(f.meta.Res.Files) > 0 { + f.meta.Res.Files[0].Size = 0 + } + f.sequentialSizeUnknown = true + f.ifRange = "" + f.rangeValidatorPinned = false + f.connMu.Unlock() + 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) { @@ -1590,6 +1797,34 @@ func (f *Fetcher) resetConnectionForRestart(conn *connection) { conn.lastSpeedDownload = 0 } +// hasSequentialPrefixLocked reports whether a single sequential connection has +// a useful contiguous prefix. The caller must hold connMu. +func (f *Fetcher) hasSequentialPrefixLocked(conn *connection) bool { + if f.meta == nil || f.meta.Res == nil || f.meta.Res.Range || !f.rangeReprobeEligible || f.meta.Res.Size <= 0 { + return false + } + if conn == nil || conn.ID != 0 || len(f.connections) != 1 || conn.Chunk == nil { + return false + } + downloaded := conn.Chunk.Downloaded + return downloaded > 0 && downloaded < f.meta.Res.Size +} + +// canProbeSequentialResumeLocked reports whether the prefix can be resumed +// safely with If-Range. The caller must hold connMu. +func (f *Fetcher) canProbeSequentialResumeLocked(conn *connection) bool { + return !f.sequentialSizeUnknown && f.ifRange != "" && f.hasSequentialPrefixLocked(conn) +} + +func (f *Fetcher) hasSequentialRecoveryCandidate() bool { + f.connMu.Lock() + defer f.connMu.Unlock() + if len(f.connections) != 1 { + return false + } + return f.hasSequentialPrefixLocked(f.connections[0]) +} + func (f *Fetcher) resumeConnections() { // Collect connections to resume while holding the lock var toResume []*connection @@ -1613,7 +1848,9 @@ func (f *Fetcher) resumeConnections() { continue } } - f.resetConnectionForRestart(conn) + if !f.hasSequentialPrefixLocked(conn) { + f.resetConnectionForRestart(conn) + } // Reset the connection state for resume conn.ctx, conn.cancel = context.WithCancel(f.ctx) conn.State = connNotStarted @@ -1683,7 +1920,7 @@ func (f *Fetcher) onDownloadComplete() { // may be detected only after the expected bytes have already been written. var finalErr error for _, conn := range f.connections { - if conn.State == connFailed && conn.failed && isRangeIntegrityError(conn.lastErr) { + if conn.State == connFailed && conn.failed && isTerminalRangeError(conn.lastErr) { finalErr = fmt.Errorf("connection %d failed: retries=%d, err=%w", conn.ID, conn.retryTimes, conn.lastErr) break } diff --git a/internal/protocol/http/fetcher_manager.go b/internal/protocol/http/fetcher_manager.go index 22666553..baf7d21e 100644 --- a/internal/protocol/http/fetcher_manager.go +++ b/internal/protocol/http/fetcher_manager.go @@ -14,8 +14,15 @@ import ( // ============================================================================ type fetcherData struct { - Connections []*connection - RedirectURL string // Saved redirect URL for resume + Connections []*connection + RedirectURL string // Saved redirect URL for resume + IfRange string // Strong ETag or Last-Modified validator for safe resume + RangeReprobeEligible bool // Origin advertised Range before falling back to sequential mode + RangeValidatorPinned bool // Recovered Range mode must keep the same If-Range validator + SequentialSizeUnknown bool // Interrupted chunked restart must retry from byte zero + Range *bool // Authoritative Range mode; nil for records saved by older versions + ResourceSize *int64 // Authoritative resource size from the same snapshot as Connections + FileSize *int64 // Authoritative single-file size; nil for older records } // ============================================================================ @@ -75,12 +82,69 @@ func (fm *FetcherManager) Store(f fetcher.Fetcher) (data any, err error) { _f.redirectLock.Lock() redirectURL := _f.redirectURL _f.redirectLock.Unlock() + + // Build an immutable snapshot while holding the same lock used by download + // workers to update Range mode and connection progress. The storage layer + // marshals this value after Store returns, so returning live pointers would + // otherwise race with the active download and could persist a mixed state. + _f.connMu.Lock() + connections := snapshotConnectionsLocked(_f.connections) + ifRange := _f.ifRange + rangeReprobeEligible := _f.rangeReprobeEligible + rangeValidatorPinned := _f.rangeValidatorPinned + sequentialSizeUnknown := _f.sequentialSizeUnknown + var rangeMode *bool + var resourceSize *int64 + var fileSize *int64 + if _f.meta != nil && _f.meta.Res != nil { + value := _f.meta.Res.Range + rangeMode = &value + size := _f.meta.Res.Size + resourceSize = &size + if len(_f.meta.Res.Files) > 0 && _f.meta.Res.Files[0] != nil { + size := _f.meta.Res.Files[0].Size + fileSize = &size + } + } + _f.connMu.Unlock() + return &fetcherData{ - Connections: _f.connections, - RedirectURL: redirectURL, + Connections: connections, + RedirectURL: redirectURL, + IfRange: ifRange, + RangeReprobeEligible: rangeReprobeEligible, + RangeValidatorPinned: rangeValidatorPinned, + SequentialSizeUnknown: sequentialSizeUnknown, + Range: rangeMode, + ResourceSize: resourceSize, + FileSize: fileSize, }, nil } +// snapshotConnectionsLocked copies only the serialized connection state and +// deep-copies chunks. The caller must hold the fetcher's connMu. +func snapshotConnectionsLocked(connections []*connection) []*connection { + snapshot := make([]*connection, len(connections)) + for i, conn := range connections { + if conn == nil { + continue + } + copyConn := &connection{ + ID: conn.ID, + Role: conn.Role, + State: conn.State, + Downloaded: conn.Downloaded, + Completed: conn.Completed, + } + if conn.Chunk != nil { + copyChunk := *conn.Chunk + copyConn.Chunk = ©Chunk + } + snapshot[i] = copyConn + } + return snapshot +} + func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) { return &fetcherData{}, func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher { fd := v.(*fetcherData) @@ -96,6 +160,22 @@ func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v if fd.RedirectURL != "" { fetcher.redirectURL = fd.RedirectURL } + fetcher.ifRange = fd.IfRange + fetcher.rangeReprobeEligible = fd.RangeReprobeEligible + fetcher.rangeValidatorPinned = fd.RangeValidatorPinned + fetcher.sequentialSizeUnknown = fd.SequentialSizeUnknown + if fd.Range != nil && fetcher.meta.Res != nil { + // Recovery-critical resource fields live in the same persisted snapshot + // as Connections, making them authoritative over task metadata that may + // have been saved just before or after this record. + fetcher.meta.Res.Range = *fd.Range + } + if fd.ResourceSize != nil && fetcher.meta.Res != nil { + fetcher.meta.Res.Size = *fd.ResourceSize + } + if fd.FileSize != nil && fetcher.meta.Res != nil && len(fetcher.meta.Res.Files) > 0 && fetcher.meta.Res.Files[0] != nil { + fetcher.meta.Res.Files[0].Size = *fd.FileSize + } return fetcher } } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index ef8723d0..8b2f66c5 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -529,6 +529,537 @@ func TestFetcher_DownloadIgnoredRangeFallsBackToSequential(t *testing.T) { } } +func TestFetcher_SequentialRetryReprobesRange(t *testing.T) { + payload := make([]byte, 4*1024*1024) + for i := range payload { + payload[i] = byte(i % 251) + } + shortReplacement := make([]byte, 2*1024*1024) + chunkedReplacement := make([]byte, 5*1024*1024) + for i := range shortReplacement { + shortReplacement[i] = byte((i*3 + 17) % 251) + } + for i := range chunkedReplacement { + chunkedReplacement[i] = byte((i*7 + 29) % 251) + } + + const ( + etagA = `"range-reprobe-a"` + etagB = `"range-reprobe-b"` + weakETag = `W/"range-reprobe-b"` + lastModified = "Wed, 21 Oct 2015 07:28:00 GMT" + laterDate = "Wed, 21 Oct 2015 07:28:02 GMT" + failedSize = 128 * 1024 + ) + + tests := []struct { + name string + resolveETag string + ignoredETag string + lastModified string + date string + wantIfRange string + resumeStatus int + wantRangeMode bool + wantParallel bool + wantRangeRequest int32 + wantFullRestart bool + replacement []byte + fullRestartReplacement []byte + chunked200 bool + }{ + { + name: "ignored response replaces Resolve ETag", + resolveETag: etagA, + ignoredETag: etagB, + wantIfRange: etagB, + resumeStatus: gohttp.StatusPartialContent, + wantRangeMode: true, + wantParallel: true, + wantRangeRequest: 2, + }, + { + name: "strong Last-Modified resumes with 206", + lastModified: lastModified, + date: laterDate, + wantIfRange: lastModified, + resumeStatus: gohttp.StatusPartialContent, + wantRangeMode: true, + wantParallel: true, + wantRangeRequest: 2, + }, + { + name: "missing validator restarts from zero", + resumeStatus: gohttp.StatusPartialContent, + wantRangeRequest: 1, + wantFullRestart: true, + fullRestartReplacement: shortReplacement, + }, + { + name: "weak ETag blocks Last-Modified fallback", + ignoredETag: weakETag, + lastModified: lastModified, + date: laterDate, + resumeStatus: gohttp.StatusPartialContent, + wantRangeRequest: 1, + wantFullRestart: true, + }, + { + name: "same-second Last-Modified restarts from zero", + lastModified: lastModified, + date: lastModified, + resumeStatus: gohttp.StatusPartialContent, + wantRangeRequest: 1, + wantFullRestart: true, + }, + { + name: "guarded probe returning shorter 200 replaces resource", + ignoredETag: etagB, + wantIfRange: etagB, + resumeStatus: gohttp.StatusOK, + wantRangeRequest: 2, + replacement: shortReplacement, + }, + { + name: "guarded probe returning larger chunked 200 replaces resource", + ignoredETag: etagB, + wantIfRange: etagB, + resumeStatus: gohttp.StatusOK, + wantRangeRequest: 2, + replacement: chunkedReplacement, + chunked200: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var rangeRequests atomic.Int32 + var fullRequests atomic.Int32 + var resumedAt atomic.Int64 + var sawIfRange atomic.Bool + var workerValidatorMismatch atomic.Bool + + 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 == "" { + requestNumber := fullRequests.Add(1) + responsePayload := payload + if requestNumber > 1 && tt.fullRestartReplacement != nil { + responsePayload = tt.fullRestartReplacement + } + if requestNumber == 1 && tt.resolveETag != "" { + w.Header().Set(base.HttpHeaderETag, tt.resolveETag) + } + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(responsePayload))) + w.WriteHeader(gohttp.StatusOK) + if requestNumber == 1 { + if flusher, ok := w.(gohttp.Flusher); ok { + 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 + } + _, _ = w.Write(responsePayload) + return + } + + requestNumber := rangeRequests.Add(1) + if requestNumber == 1 { + // Simulate the real origin: it first ignores Range, then drops the + // sequential response after a useful prefix has been written. + if tt.ignoredETag != "" { + w.Header().Set(base.HttpHeaderETag, tt.ignoredETag) + } + if tt.lastModified != "" { + w.Header().Set(base.HttpHeaderLastModified, tt.lastModified) + } + if tt.date != "" { + w.Header().Set("Date", tt.date) + } + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(payload))) + w.WriteHeader(gohttp.StatusOK) + if flusher, ok := w.(gohttp.Flusher); ok { + flusher.Flush() + } + _, _ = w.Write(payload[:failedSize]) + return + } + + var start, end int64 + if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end); err != nil { + w.WriteHeader(gohttp.StatusBadRequest) + return + } + if requestNumber == 2 { + resumedAt.Store(start) + sawIfRange.Store(r.Header.Get(base.HttpHeaderIfRange) == tt.wantIfRange) + } else if tt.wantParallel && r.Header.Get(base.HttpHeaderIfRange) != tt.wantIfRange { + workerValidatorMismatch.Store(true) + } + if requestNumber == 2 && tt.resumeStatus == gohttp.StatusOK { + responsePayload := payload + if tt.replacement != nil { + responsePayload = tt.replacement + } + w.Header().Set(base.HttpHeaderETag, `"replacement"`) + if !tt.chunked200 { + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(responsePayload))) + } + w.WriteHeader(gohttp.StatusOK) + if tt.chunked200 { + w.(gohttp.Flusher).Flush() + } + for offset := 0; offset < len(responsePayload); offset += 8 * 1024 { + end := min(offset+8*1024, len(responsePayload)) + if _, err := w.Write(responsePayload[offset:end]); err != nil { + return + } + } + 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) + if requestNumber == 2 && tt.wantParallel { + w.(gohttp.Flusher).Flush() + deadline := time.Now().Add(2 * time.Second) + for rangeRequests.Load() < 3 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + } + 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 + } + time.Sleep(time.Millisecond) + } + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/dynamic-range.data"}, &base.Options{ + Path: t.TempDir(), + Name: "dynamic-range.data", + Extra: &http.OptsExtra{ + Connections: 4, + }, + }); 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 < tt.wantRangeRequest { + t.Fatalf("Range requests = %d, want at least %d", got, tt.wantRangeRequest) + } + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if tt.wantFullRestart && fullRequests.Load() < 2 { + t.Fatal("missing validator did not issue a full restart request") + } + if tt.wantIfRange != "" && resumedAt.Load() != failedSize { + t.Fatalf("resume Range started at %d, want %d", resumedAt.Load(), failedSize) + } + if tt.wantIfRange != "" && !sawIfRange.Load() { + t.Fatalf("resume probe did not include If-Range %q", tt.wantIfRange) + } + if f.meta.Res.Range != tt.wantRangeMode { + t.Fatalf("Range mode = %v, want %v", f.meta.Res.Range, tt.wantRangeMode) + } + if tt.wantParallel { + stats := f.Stats().(*http.Stats) + if len(stats.Connections) <= 1 { + f.slowStart.mu.Lock() + totalLaunched := f.slowStart.totalLaunched + batchPending := f.slowStart.batchPending + batchReady := f.slowStart.batchReady + maxConnections := f.slowStart.maxConnections + nextBatchSize := f.slowStart.nextBatchSize + paused := f.slowStart.paused + f.slowStart.mu.Unlock() + t.Fatalf("successful re-probe kept %d connection, want expansion (requests=%d state=%d launched=%d pending=%d ready=%d max=%d next=%d paused=%v)", len(stats.Connections), rangeRequests.Load(), f.getState(), totalLaunched, batchPending, batchReady, maxConnections, nextBatchSize, paused) + } + if workerValidatorMismatch.Load() { + t.Fatalf("expanded Range worker omitted pinned If-Range %q", tt.wantIfRange) + } + } + wantPayload := payload + if tt.replacement != nil { + wantPayload = tt.replacement + } else if tt.fullRestartReplacement != nil { + wantPayload = tt.fullRestartReplacement + } + if !bytes.Equal(got, wantPayload) { + t.Fatal("downloaded file differs after sequential recovery") + } + if f.meta.Res.Size != int64(len(wantPayload)) || f.meta.Res.Files[0].Size != int64(len(wantPayload)) { + t.Fatalf("resource size = %d/%d, want %d", f.meta.Res.Size, f.meta.Res.Files[0].Size, len(wantPayload)) + } + if got := f.Progress().TotalDownloaded(); got != int64(len(wantPayload)) { + t.Fatalf("progress = %d, want %d", got, len(wantPayload)) + } + info, err := os.Stat(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if got := info.Size(); got != int64(len(wantPayload)) { + t.Fatalf("file size = %d, want %d", got, len(wantPayload)) + } + }) + } +} + +func TestFetcher_InitialIgnoredRangeUsesResponseSize(t *testing.T) { + resolvePayload := bytes.Repeat([]byte("resolve"), 64*1024) + replacement := bytes.Repeat([]byte("replacement"), 24*1024) + 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.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(resolvePayload))) + w.WriteHeader(gohttp.StatusOK) + _, _ = w.Write(resolvePayload) + return + } + + rangeRequests.Add(1) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(replacement))) + w.WriteHeader(gohttp.StatusOK) + _, _ = w.Write(replacement) + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/ignored-range.data"}, &base.Options{ + Path: t.TempDir(), + Name: "ignored-range.data", + Extra: &http.OptsExtra{ + Connections: 4, + }, + }); err != nil { + t.Fatal(err) + } + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, replacement) { + t.Fatal("initial ignored-Range response did not replace the resolved representation") + } + if rangeRequests.Load() != 1 { + t.Fatalf("Range requests = %d, want 1", rangeRequests.Load()) + } + wantSize := int64(len(replacement)) + if f.meta.Res.Size != wantSize || f.meta.Res.Files[0].Size != wantSize { + t.Fatalf("resource size = %d/%d, want %d", f.meta.Res.Size, f.meta.Res.Files[0].Size, wantSize) + } + if got := f.Progress().TotalDownloaded(); got != wantSize { + t.Fatalf("progress = %d, want %d", got, wantSize) + } + info, err := os.Stat(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if got := info.Size(); got != wantSize { + t.Fatalf("file size = %d, want %d", got, wantSize) + } +} + +func TestFetcher_InterruptedChunkedReplacementRestartsFromZero(t *testing.T) { + payload := bytes.Repeat([]byte("original"), 512*1024) + replacement := bytes.Repeat([]byte("replacement"), 256*1024) + const firstPrefix = 128 * 1024 + const replacementPrefix = 96 * 1024 + var rangeRequests atomic.Int32 + var fullRequests atomic.Int32 + var sawFullRestart atomic.Bool + + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { + w.Header().Set(base.HttpHeaderAcceptRanges, base.HttpHeaderBytes) + rangeHeader := r.Header.Get(base.HttpHeaderRange) + if rangeHeader == "" { + requestNumber := fullRequests.Add(1) + responsePayload := payload + if requestNumber > 1 { + responsePayload = replacement + sawFullRestart.Store(true) + } + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(responsePayload))) + w.WriteHeader(gohttp.StatusOK) + _, _ = w.Write(responsePayload) + return + } + + requestNumber := rangeRequests.Add(1) + switch requestNumber { + case 1: + w.Header().Set(base.HttpHeaderETag, `"prefix-v1"`) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(payload))) + w.WriteHeader(gohttp.StatusOK) + w.(gohttp.Flusher).Flush() + _, _ = w.Write(payload[:firstPrefix]) + case 2: + if got := r.Header.Get(base.HttpHeaderIfRange); got != `"prefix-v1"` { + t.Errorf("If-Range = %q, want %q", got, `"prefix-v1"`) + } + conn, rw, err := w.(gohttp.Hijacker).Hijack() + if err != nil { + t.Errorf("Hijack() error = %v", err) + return + } + defer conn.Close() + _, _ = fmt.Fprintf(rw, "HTTP/1.1 200 OK\r\nETag: \"replacement-v2\"\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n%x\r\n", replacementPrefix) + _, _ = rw.Write(replacement[:replacementPrefix]) + _, _ = rw.WriteString("\r\n") + _ = rw.Flush() + default: + t.Errorf("unexpected Range retry after unknown-size replacement: %q", rangeHeader) + w.WriteHeader(gohttp.StatusBadRequest) + } + })) + defer server.Close() + + f := buildFetcher() + if err := f.Resolve(&base.Request{URL: server.URL + "/chunked-replacement.data"}, &base.Options{ + Path: t.TempDir(), + Name: "chunked-replacement.data", + Extra: &http.OptsExtra{ + Connections: 4, + }, + }); err != nil { + t.Fatal(err) + } + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + + got, err := os.ReadFile(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, replacement) { + t.Fatal("downloaded file differs after interrupted chunked replacement") + } + if rangeRequests.Load() != 2 { + t.Fatalf("Range requests = %d, want 2", rangeRequests.Load()) + } + if !sawFullRestart.Load() { + t.Fatal("interrupted chunked replacement did not restart with a full request") + } + wantSize := int64(len(replacement)) + if f.meta.Res.Size != wantSize || f.meta.Res.Files[0].Size != wantSize { + t.Fatalf("resource size = %d/%d, want %d", f.meta.Res.Size, f.meta.Res.Files[0].Size, wantSize) + } + if got := f.Progress().TotalDownloaded(); got != wantSize { + t.Fatalf("progress = %d, want %d", got, wantSize) + } + info, err := os.Stat(f.meta.SingleFilepath()) + if err != nil { + t.Fatal(err) + } + if got := info.Size(); got != wantSize { + t.Fatalf("file size = %d, want %d", got, wantSize) + } + if f.sequentialSizeUnknown { + t.Fatal("completed replacement kept unknown-size recovery state") + } +} + +func TestExtractIfRangeValidator(t *testing.T) { + const ( + lastModified = "Wed, 21 Oct 2015 07:28:00 GMT" + laterDate = "Wed, 21 Oct 2015 07:28:02 GMT" + ) + tests := []struct { + name string + etag string + lm string + date string + want string + }{ + {name: "strong ETag", etag: `"v1"`, want: `"v1"`}, + {name: "malformed ETag", etag: `"bad value"`}, + {name: "weak ETag blocks date", etag: `W/"v1"`, lm: lastModified, date: laterDate}, + {name: "strong Last-Modified", lm: lastModified, date: laterDate, want: lastModified}, + {name: "same-second Last-Modified", lm: lastModified, date: lastModified}, + {name: "Last-Modified without Date", lm: lastModified}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + header := make(gohttp.Header) + if tt.etag != "" { + header.Set(base.HttpHeaderETag, tt.etag) + } + if tt.lm != "" { + header.Set(base.HttpHeaderLastModified, tt.lm) + } + if tt.date != "" { + header.Set("Date", tt.date) + } + if got := extractIfRangeValidator(header); got != tt.want { + t.Fatalf("extractIfRangeValidator() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestFetcher_OriginalURLFallbackPreservesResumeHeaders(t *testing.T) { + const ( + start = int64(128) + end = int64(255) + ifRange = `"resume-v2"` + ) + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { + if got, want := r.Header.Get(base.HttpHeaderRange), "bytes=128-255"; got != want { + t.Errorf("Range = %q, want %q", got, want) + } + if got := r.Header.Get(base.HttpHeaderIfRange); got != ifRange { + t.Errorf("If-Range = %q, want %q", got, ifRange) + } + w.Header().Set(base.HttpHeaderContentRange, "bytes 128-255/256") + w.WriteHeader(gohttp.StatusPartialContent) + })) + defer server.Close() + + f := &Fetcher{ + config: &config{}, + meta: &fetcher.FetcherMeta{Req: &base.Request{URL: server.URL}}, + } + resp, err := f.tryFallbackToOriginalURL(context.Background(), server.Client(), start, end, true, ifRange) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() +} + func TestFetcher_RejectsInvalidPartialContent(t *testing.T) { const ( requestedStart = int64(1024) @@ -702,6 +1233,7 @@ func TestFetcher_InvalidResponseLengthFailsTask(t *testing.T) { bodyChunks []int resolvedDownloaded int64 wantDownloaded int64 + rangeMode bool }{ { name: "sequential declared short body", @@ -732,14 +1264,19 @@ func TestFetcher_InvalidResponseLengthFailsTask(t *testing.T) { bodyChunks: []int{int(rangeLength), 1}, resolvedDownloaded: requestedStart, wantDownloaded: rangeLength, + rangeMode: 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-4095"; got != want { - t.Errorf("Range header = %q, want %q", got, want) + wantRange := "" + if tc.rangeMode { + wantRange = "bytes=1024-4095" + } + if got := request.Header.Get(base.HttpHeaderRange); got != wantRange { + t.Errorf("Range header = %q, want %q", got, wantRange) } if tc.contentRange != "" { writer.Header().Set(base.HttpHeaderContentRange, tc.contentRange) @@ -761,7 +1298,7 @@ func TestFetcher_InvalidResponseLengthFailsTask(t *testing.T) { f := buildFetcher() f.meta.Req = &base.Request{URL: server.URL} - f.meta.Res = &base.Resource{Range: true, Size: totalSize} + f.meta.Res = &base.Resource{Range: tc.rangeMode, Size: totalSize} output, err := os.CreateTemp(t.TempDir(), "range-task-*") if err != nil { t.Fatal(err) @@ -771,11 +1308,15 @@ func TestFetcher_InvalidResponseLengthFailsTask(t *testing.T) { } f.file = output + connChunk := newChunk(0, totalSize-1) + if tc.rangeMode { + connChunk = newChunk(requestedStart, requestedEnd) + } conn := &connection{ ID: 0, Role: rolePrimary, State: connNotStarted, - Chunk: newChunk(requestedStart, requestedEnd), + Chunk: connChunk, ctx: context.Background(), } f.connections = []*connection{conn} @@ -1505,6 +2046,308 @@ func TestFetcherManager_ParseName(t *testing.T) { } } +func TestFetcherManager_StoreCreatesAtomicSnapshot(t *testing.T) { + f := &Fetcher{ + meta: &fetcher.FetcherMeta{ + Req: &base.Request{}, + Res: &base.Resource{ + Size: 100, + Range: true, + Files: []*base.FileInfo{{Size: 100}}, + }, + Opts: &base.Options{}, + }, + ifRange: `"snapshot-v1"`, + rangeValidatorPinned: true, + connections: []*connection{{ + ID: 0, + Role: rolePrimary, + State: connDownloading, + Chunk: newChunk(0, 99), + }}, + } + fm := new(FetcherManager) + + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + sequential := false + for { + select { + case <-stop: + return + default: + } + f.connMu.Lock() + if sequential { + f.meta.Res.Range = false + f.meta.Res.Size = 0 + f.meta.Res.Files[0].Size = 0 + f.rangeReprobeEligible = true + f.rangeValidatorPinned = false + f.sequentialSizeUnknown = true + f.connections[0].Downloaded = 64 + f.connections[0].Chunk.Downloaded = 64 + } else { + f.meta.Res.Range = true + f.meta.Res.Size = 100 + f.meta.Res.Files[0].Size = 100 + f.rangeReprobeEligible = false + f.rangeValidatorPinned = true + f.sequentialSizeUnknown = false + f.connections[0].Downloaded = 0 + f.connections[0].Chunk.Downloaded = 0 + } + f.connMu.Unlock() + sequential = !sequential + } + }() + + for i := 0; i < 1000; i++ { + data, err := fm.Store(f) + if err != nil { + close(stop) + <-done + t.Fatal(err) + } + fd := data.(*fetcherData) + if fd.Range == nil { + close(stop) + <-done + t.Fatal("snapshot Range mode is nil") + } + if fd.ResourceSize == nil || fd.FileSize == nil { + close(stop) + <-done + t.Fatal("snapshot sizes are nil") + } + if len(fd.Connections) != 1 || fd.Connections[0] == nil || fd.Connections[0].Chunk == nil { + close(stop) + <-done + t.Fatalf("invalid connection snapshot: %#v", fd.Connections) + } + if fd.Connections[0] == f.connections[0] || fd.Connections[0].Chunk == f.connections[0].Chunk { + close(stop) + <-done + t.Fatal("Store returned live connection state instead of a deep snapshot") + } + + downloaded := fd.Connections[0].Chunk.Downloaded + if *fd.Range { + if fd.RangeReprobeEligible || !fd.RangeValidatorPinned || fd.SequentialSizeUnknown || downloaded != 0 || *fd.ResourceSize != 100 || *fd.FileSize != 100 { + close(stop) + <-done + t.Fatalf("mixed ranged snapshot: eligible=%v pinned=%v unknown=%v downloaded=%d size=%d/%d", fd.RangeReprobeEligible, fd.RangeValidatorPinned, fd.SequentialSizeUnknown, downloaded, *fd.ResourceSize, *fd.FileSize) + } + } else if !fd.RangeReprobeEligible || fd.RangeValidatorPinned || !fd.SequentialSizeUnknown || downloaded != 64 || *fd.ResourceSize != 0 || *fd.FileSize != 0 { + close(stop) + <-done + t.Fatalf("mixed sequential snapshot: eligible=%v pinned=%v unknown=%v downloaded=%d size=%d/%d", fd.RangeReprobeEligible, fd.RangeValidatorPinned, fd.SequentialSizeUnknown, downloaded, *fd.ResourceSize, *fd.FileSize) + } + } + close(stop) + <-done +} + +func TestFetcherManager_RestoreUsesSnapshotRangeMode(t *testing.T) { + fm := new(FetcherManager) + rangeMode := false + resourceSize := int64(0) + fileSize := int64(0) + saved := &fetcherData{ + Connections: []*connection{{ + ID: 0, + Role: rolePrimary, + State: connFailed, + Chunk: &chunk{Begin: 0, End: 99, Downloaded: 64}, + Downloaded: 64, + }}, + IfRange: `"snapshot-v1"`, + RangeReprobeEligible: true, + SequentialSizeUnknown: true, + Range: &rangeMode, + ResourceSize: &resourceSize, + FileSize: &fileSize, + } + encoded, err := json.Marshal(saved) + if err != nil { + t.Fatal(err) + } + var persisted fetcherData + if err := json.Unmarshal(encoded, &persisted); err != nil { + t.Fatal(err) + } + + staleMeta := &fetcher.FetcherMeta{ + Req: &base.Request{}, + Res: &base.Resource{ + Size: 100, + Range: true, + Files: []*base.FileInfo{{Size: 100}}, + }, + Opts: &base.Options{}, + } + _, restore := fm.Restore() + restored := restore(staleMeta, &persisted).(*Fetcher) + if restored.meta.Res.Range { + t.Fatal("Restore kept stale task Range mode instead of the connection snapshot mode") + } + if !restored.rangeReprobeEligible || !restored.sequentialSizeUnknown || restored.ifRange != `"snapshot-v1"` { + t.Fatalf("restored recovery state = eligible:%v unknown:%v If-Range:%q", restored.rangeReprobeEligible, restored.sequentialSizeUnknown, restored.ifRange) + } + if restored.meta.Res.Size != 0 || restored.meta.Res.Files[0].Size != 0 { + t.Fatalf("restored snapshot size = %d/%d, want 0/0", restored.meta.Res.Size, restored.meta.Res.Files[0].Size) + } + if got := restored.connections[0].Chunk.Downloaded; got != 64 { + t.Fatalf("restored prefix = %d, want 64", got) + } + restored.connMu.Lock() + canProbe := restored.canProbeSequentialResumeLocked(restored.connections[0]) + restored.connMu.Unlock() + if canProbe { + t.Fatal("unknown-size restored state attempted an If-Range resume against stale size") + } + + pinnedRangeMode := true + pinnedMeta := &fetcher.FetcherMeta{ + Req: &base.Request{}, + Res: &base.Resource{Size: 100, Range: false}, + Opts: &base.Options{}, + } + pinned := restore(pinnedMeta, &fetcherData{ + IfRange: `"snapshot-v2"`, + RangeValidatorPinned: true, + Range: &pinnedRangeMode, + }).(*Fetcher) + if !pinned.meta.Res.Range || !pinned.rangeValidatorPinned || pinned.ifRange != `"snapshot-v2"` { + t.Fatalf("restored pinned range state = range:%v pinned:%v If-Range:%q", pinned.meta.Res.Range, pinned.rangeValidatorPinned, pinned.ifRange) + } + + // Records written before the Range snapshot field existed must retain the + // mode from task metadata for backward compatibility. + legacyMeta := &fetcher.FetcherMeta{ + Req: &base.Request{}, + Res: &base.Resource{Size: 100, Range: true}, + Opts: &base.Options{}, + } + legacy := restore(legacyMeta, &fetcherData{}).(*Fetcher) + if !legacy.meta.Res.Range { + t.Fatal("legacy record unexpectedly overrode task Range mode") + } +} + +func TestFetcherManager_RestoreUsesAtomicSizeSnapshotForDownload(t *testing.T) { + original := bytes.Repeat([]byte("old-representation"), 16*1024) + replacement := bytes.Repeat([]byte("new-representation"), 12*1024) + const savedPrefix = 64 * 1024 + const replacementPrefix = 96 * 1024 + var sawGuardedProbe atomic.Bool + + server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { + wantRange := fmt.Sprintf("bytes=%d-%d", savedPrefix, len(original)-1) + if got := r.Header.Get(base.HttpHeaderRange); got != wantRange { + t.Errorf("Range = %q, want %q", got, wantRange) + } + if got := r.Header.Get(base.HttpHeaderIfRange); got != `"old-v1"` { + t.Errorf("If-Range = %q, want %q", got, `"old-v1"`) + } + sawGuardedProbe.Store(true) + w.Header().Set(base.HttpHeaderETag, `"new-v2"`) + w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(replacement))) + w.WriteHeader(gohttp.StatusOK) + _, _ = w.Write(replacement) + })) + defer server.Close() + + tempDir := t.TempDir() + const filename = "restored-replacement.data" + filepath := tempDir + string(os.PathSeparator) + filename + if err := os.WriteFile(filepath, replacement[:replacementPrefix], 0o644); err != nil { + t.Fatal(err) + } + + rangeMode := false + resourceSize := int64(len(original)) + fileSize := resourceSize + saved := &fetcherData{ + Connections: []*connection{{ + ID: 0, + Role: rolePrimary, + State: connFailed, + Chunk: &chunk{Begin: 0, End: resourceSize - 1, Downloaded: savedPrefix}, + Downloaded: savedPrefix, + }}, + IfRange: `"old-v1"`, + RangeReprobeEligible: true, + Range: &rangeMode, + ResourceSize: &resourceSize, + FileSize: &fileSize, + } + // Simulate task metadata serialized after an interrupted chunked replacement + // while the fetcher snapshot still describes the preceding representation. + staleMeta := &fetcher.FetcherMeta{ + Req: &base.Request{URL: server.URL}, + Res: &base.Resource{ + Size: 0, + Range: false, + Files: []*base.FileInfo{{Name: filename, Size: 0}}, + }, + Opts: &base.Options{ + Path: tempDir, + Name: filename, + Extra: &http.OptsExtra{ + Connections: 1, + }, + }, + } + + fm := new(FetcherManager) + _, restore := fm.Restore() + restored := restore(staleMeta, saved).(*Fetcher) + if restored.meta.Res.Size != resourceSize || restored.meta.Res.Files[0].Size != fileSize { + t.Fatalf("restored size = %d/%d, want %d/%d", restored.meta.Res.Size, restored.meta.Res.Files[0].Size, resourceSize, fileSize) + } + ctl := controller.NewController() + ctl.GetConfig = func(v any) { + if err := json.Unmarshal([]byte(test.ToJson(fm.DefaultConfig())), v); err != nil { + t.Fatal(err) + } + } + restored.Setup(ctl) + if err := restored.Start(); err != nil { + t.Fatal(err) + } + if err := restored.Wait(); err != nil { + t.Fatal(err) + } + + if !sawGuardedProbe.Load() { + t.Fatal("restored download did not issue the guarded Range probe") + } + got, err := os.ReadFile(filepath) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, replacement) { + t.Fatal("restored download retained bytes from the interrupted replacement") + } + wantSize := int64(len(replacement)) + if restored.meta.Res.Size != wantSize || restored.meta.Res.Files[0].Size != wantSize { + t.Fatalf("completed size = %d/%d, want %d", restored.meta.Res.Size, restored.meta.Res.Files[0].Size, wantSize) + } + if got := restored.Progress().TotalDownloaded(); got != wantSize { + t.Fatalf("progress = %d, want %d", got, wantSize) + } + info, err := os.Stat(filepath) + if err != nil { + t.Fatal(err) + } + if got := info.Size(); got != wantSize { + t.Fatalf("file size = %d, want %d", got, wantSize) + } +} + func downloadReady(listener net.Listener, connections int, t *testing.T) fetcher.Fetcher { return doDownloadReady(buildFetcher(), listener, connections, t) } diff --git a/internal/protocol/http/helper.go b/internal/protocol/http/helper.go index ae0395e1..1b796a62 100644 --- a/internal/protocol/http/helper.go +++ b/internal/protocol/http/helper.go @@ -27,6 +27,45 @@ type RequestError struct { Code int } +// extractIfRangeValidator returns only validators that are strong enough for +// If-Range. Any entity-tag header blocks Last-Modified fallback: a weak or +// malformed ETag does not make the date validator strong. +func extractIfRangeValidator(header http.Header) string { + etag := strings.TrimSpace(header.Get(base.HttpHeaderETag)) + if etag != "" { + if isStrongETag(etag) { + return etag + } + return "" + } + + lastModified := strings.TrimSpace(header.Get(base.HttpHeaderLastModified)) + date := strings.TrimSpace(header.Get("Date")) + if lastModified == "" || date == "" { + return "" + } + modifiedAt, modifiedErr := http.ParseTime(lastModified) + responseAt, responseErr := http.ParseTime(date) + if modifiedErr != nil || responseErr != nil || responseAt.Sub(modifiedAt) < time.Second { + return "" + } + return lastModified +} + +func isStrongETag(value string) bool { + if len(value) < 2 || value[0] != '"' || value[len(value)-1] != '"' { + return false + } + for i := 1; i < len(value)-1; i++ { + b := value[i] + if b == 0x21 || (b >= 0x23 && b <= 0x7e) || b >= 0x80 { + continue + } + return false + } + return true +} + func NewRequestError(code int) *RequestError { return &RequestError{Code: code} } @@ -166,15 +205,18 @@ 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) (*http.Response, error) { +func (f *Fetcher) tryFallbackToOriginalURL(ctx context.Context, client *http.Client, rangeStart, rangeEnd int64, rangeRequested bool, ifRange string) (*http.Response, error) { httpReq, err := f.buildRequestWithOriginalURL(ctx, f.meta.Req) if err != nil { return nil, err } - if f.meta.Res.Range && rangeEnd > 0 { + if rangeRequested { httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, rangeStart, rangeEnd)) + if ifRange != "" { + httpReq.Header.Set(base.HttpHeaderIfRange, ifRange) + } } resp, err := client.Do(httpReq) diff --git a/pkg/base/constants.go b/pkg/base/constants.go index d9bdb5a8..c57234fa 100644 --- a/pkg/base/constants.go +++ b/pkg/base/constants.go @@ -23,6 +23,8 @@ const ( HttpHeaderContentDisposition = "Content-Disposition" HttpHeaderUserAgent = "User-Agent" HttpHeaderLastModified = "Last-Modified" + HttpHeaderETag = "ETag" + HttpHeaderIfRange = "If-Range" HttpHeaderBytes = "bytes" HttpHeaderRangeFormat = "bytes=%d-%d"