fix(http): re-probe range support after sequential retry (#1464)

This commit is contained in:
ggg authored and GitHub committed 2026-09-02 14:29:48 +08:00
1 parent c5c8bb2bf5
commit f49e2510f5
5 files changed
+1245 -41

No files matched your search

+268 -31
View File
@@ -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
}
+84 -4
View File
@@ -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 = &copyChunk
}
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
}
}
+847 -4
View File
@@ -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)
}
+44 -2
View File
@@ -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)
+2
View File
@@ -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"