From 9adc6876e052b7665827fb27a04688573f9bcaa3 Mon Sep 17 00:00:00 2001 From: Peter Biro Date: Wed, 21 Oct 2026 21:00:58 +0200 Subject: [PATCH] fix: retry an audio chunk whose body breaks off or stalls downloadChunk retried a failed request or an unexpected status, but once the 206 came back a body that broke off or stalled until the client timeout failed the chunk, and with it the whole load. Read the body inside the same backoff op, so a broken off body is retried like any other failure, with the same 4 attempts, 1s apart. A chunk that receives no bytes for 10s is abandoned and retried instead of waiting out the 30s client timeout. The first chunk now goes through the same path. --- audio/chunked-reader.go | 121 ++++++++++++++++++-------- audio/chunked-reader_internal_test.go | 85 ++++++++++++++++++ 2 files changed, 170 insertions(+), 36 deletions(-) diff --git a/audio/chunked-reader.go b/audio/chunked-reader.go index 53228bf6..da6beec4 100644 --- a/audio/chunked-reader.go +++ b/audio/chunked-reader.go @@ -102,15 +102,13 @@ func NewHttpChunkedReader(log librespot.Logger, client *http.Client, audioUrl st }() // request the first chunk, needed for the complete content length - resp, err := r.downloadChunk(0) + firstChunk, header, err := r.downloadChunk(0) if err != nil { return nil, fmt.Errorf("failed requesting first chunk: %w", err) } - defer func() { _ = resp.Body.Close() }() - // parse the Content-Range header with the complete content length - _, _, r.len, err = parseContentRange(resp) + _, _, r.len, err = parseContentRange(&http.Response{Header: header}) if err != nil { return nil, fmt.Errorf("invalid first chunk content range response: %w", err) } @@ -123,10 +121,7 @@ func NewHttpChunkedReader(log librespot.Logger, client *http.Client, audioUrl st r.chunks[i] = newChunkItem() } - r.chunks[0].data, err = io.ReadAll(r.measureLatency(resp.Body)) - if err != nil { - return nil, fmt.Errorf("failed reading first chunk: %w", err) - } + r.chunks[0].data = firstChunk // The first chunk is fetched eagerly here, so count it towards completion. r.completedChunks = 1 @@ -147,54 +142,108 @@ func (r *HttpChunkedReader) isClosed() bool { return r.ctx.Err() != nil } -func (r *HttpChunkedReader) downloadChunk(idx int) (*http.Response, error) { +// chunkStallTimeout is how long a chunk request may go without receiving a +// byte before it is abandoned and retried. A stalled CDN connection usually +// recovers on a new request, while waiting for the client timeout holds up a +// load or seek for the whole of it. +var chunkStallTimeout = 10 * time.Second + +var errChunkStalled = errors.New("chunk download stalled") + +// downloadChunk requests and reads one chunk, returning its data and the +// response header. A failed request, an unexpected status and a body that +// breaks off or stalls are all retried. +func (r *HttpChunkedReader) downloadChunk(idx int) ([]byte, http.Header, error) { + type chunkResponse struct { + data []byte + header http.Header + } + retryBackoff := backoff.WithContext( backoff.WithMaxRetries(backoff.NewConstantBackOff(1*time.Second), 3), r.ctx, ) - return backoff.RetryWithData(func() (*http.Response, error) { - resp, err := r.client.Do((&http.Request{ - Method: "GET", - URL: r.url, - Header: http.Header{ - "User-Agent": []string{librespot.UserAgent()}, - "Range": []string{fmt.Sprintf("bytes=%d-%d", - idx*DefaultChunkSize, - min(max(r.len, DefaultChunkSize), int64((idx+1)*DefaultChunkSize))-1, - )}, - }, - }).WithContext(r.ctx)) + res, err := backoff.RetryNotifyWithData(func() (chunkResponse, error) { + data, header, err := r.downloadChunkOnce(idx) if err != nil { err = r.closeErr(err) if errors.Is(err, net.ErrClosed) { - return nil, backoff.Permanent(err) + return chunkResponse{}, backoff.Permanent(err) } - return nil, err + return chunkResponse{}, err } - if resp.StatusCode != http.StatusPartialContent { - _ = resp.Body.Close() - return nil, fmt.Errorf("unexpected chunk response status: %w", - &librespot.HTTPStatusError{Endpoint: "audio file", StatusCode: resp.StatusCode}) - } - - return resp, nil - }, retryBackoff) + return chunkResponse{data, header}, nil + }, retryBackoff, func(err error, _ time.Duration) { + r.log.Debugf("retrying chunk %d: %v", idx, err) + }) + return res.data, res.header, err } -func (r *HttpChunkedReader) downloadAndRead(idx int) ([]byte, error) { - resp, err := r.downloadChunk(idx) +func (r *HttpChunkedReader) downloadChunkOnce(idx int) ([]byte, http.Header, error) { + ctx, cancel := context.WithCancelCause(r.ctx) + defer cancel(nil) + + stall := time.AfterFunc(chunkStallTimeout, func() { cancel(errChunkStalled) }) + defer stall.Stop() + + stalled := func(err error) error { + if errors.Is(context.Cause(ctx), errChunkStalled) { + return errChunkStalled + } + return err + } + + resp, err := r.client.Do((&http.Request{ + Method: "GET", + URL: r.url, + Header: http.Header{ + "User-Agent": []string{librespot.UserAgent()}, + "Range": []string{fmt.Sprintf("bytes=%d-%d", + idx*DefaultChunkSize, + min(max(r.len, DefaultChunkSize), int64((idx+1)*DefaultChunkSize))-1, + )}, + }, + }).WithContext(ctx)) if err != nil { - return nil, fmt.Errorf("failed downloading chunk %d: %w", idx, r.closeErr(err)) + return nil, nil, stalled(err) } defer func() { _ = resp.Body.Close() }() - data, err := io.ReadAll(r.measureLatency(resp.Body)) + if resp.StatusCode != http.StatusPartialContent { + return nil, nil, fmt.Errorf("unexpected chunk response status: %w", + &librespot.HTTPStatusError{Endpoint: "audio file", StatusCode: resp.StatusCode}) + } + + data, err := io.ReadAll(r.measureLatency(&stallReader{resp.Body, stall})) if err != nil { - return nil, fmt.Errorf("failed reading chunk %d: %w", idx, r.closeErr(err)) + return nil, nil, fmt.Errorf("failed reading chunk %d: %w", idx, stalled(err)) + } + + return data, resp.Header, nil +} + +// stallReader restarts the stall timer whenever the body makes progress. +type stallReader struct { + io.Reader + timer *time.Timer +} + +func (r *stallReader) Read(p []byte) (int, error) { + n, err := r.Reader.Read(p) + if n > 0 { + r.timer.Reset(chunkStallTimeout) + } + return n, err +} + +func (r *HttpChunkedReader) downloadAndRead(idx int) ([]byte, error) { + data, _, err := r.downloadChunk(idx) + if err != nil { + return nil, fmt.Errorf("failed downloading chunk %d: %w", idx, r.closeErr(err)) } return data, nil diff --git a/audio/chunked-reader_internal_test.go b/audio/chunked-reader_internal_test.go index 7f7450ad..17b78367 100644 --- a/audio/chunked-reader_internal_test.go +++ b/audio/chunked-reader_internal_test.go @@ -246,6 +246,91 @@ func TestCloseNormalizesBodyReadTransportErrors(t *testing.T) { } } +func TestFetchChunkRetriesABrokenOffBody(t *testing.T) { + var requests int + reader := newFetchTestReader(t, roundTripperFunc(func(*http.Request) (*http.Response, error) { + requests++ + body := io.NopCloser(strings.NewReader("audio")) + if requests == 1 { + body = io.NopCloser(readerFunc(func([]byte) (int, error) { + return 0, io.ErrUnexpectedEOF + })) + } + return &http.Response{StatusCode: http.StatusPartialContent, Body: body, Header: http.Header{}}, nil + })) + + data, err := reader.fetchChunk(0) + require.NoError(t, err) + require.Equal(t, []byte("audio"), data) + require.Equal(t, 2, requests) +} + +func TestFetchChunkRetriesAStalledBody(t *testing.T) { + saved := chunkStallTimeout + chunkStallTimeout = 50 * time.Millisecond + t.Cleanup(func() { chunkStallTimeout = saved }) + + var requests int + reader := newFetchTestReader(t, roundTripperFunc(func(req *http.Request) (*http.Response, error) { + requests++ + body := io.NopCloser(strings.NewReader("audio")) + if requests == 1 { + // The body never sends a byte, like a CDN connection that hangs + // after the 206. + body = io.NopCloser(readerFunc(func([]byte) (int, error) { + <-req.Context().Done() + return 0, req.Context().Err() + })) + } + return &http.Response{StatusCode: http.StatusPartialContent, Body: body, Header: http.Header{}}, nil + })) + + data, err := reader.fetchChunk(0) + require.NoError(t, err) + require.Equal(t, []byte("audio"), data) + require.Equal(t, 2, requests) +} + +func TestFetchChunkDoesNotMultiplyStatusRetries(t *testing.T) { + var requests int + reader := newFetchTestReader(t, roundTripperFunc(func(*http.Request) (*http.Response, error) { + requests++ + return &http.Response{StatusCode: http.StatusForbidden, Body: io.NopCloser(strings.NewReader("")), Header: http.Header{}}, nil + })) + + _, err := reader.fetchChunk(0) + var status *librespot.HTTPStatusError + require.ErrorAs(t, err, &status) + require.Equal(t, http.StatusForbidden, status.StatusCode) + require.Equal(t, 4, requests) +} + +func TestNewReaderRetriesABrokenOffFirstChunk(t *testing.T) { + var requests int + client := &http.Client{Transport: roundTripperFunc(func(*http.Request) (*http.Response, error) { + requests++ + body := io.NopCloser(strings.NewReader("audio")) + if requests == 1 { + body = io.NopCloser(readerFunc(func([]byte) (int, error) { + return 0, io.ErrUnexpectedEOF + })) + } + return &http.Response{ + StatusCode: http.StatusPartialContent, + Body: body, + Header: http.Header{"Content-Range": []string{"bytes 0-4/5"}}, + }, nil + })} + + reader, err := NewHttpChunkedReader(&librespot.NullLogger{}, client, "https://example.com/audio") + require.NoError(t, err) + t.Cleanup(func() { _ = reader.Close() }) + + require.Equal(t, int64(5), reader.Size()) + require.Equal(t, []byte("audio"), reader.chunks[0].data) + require.Equal(t, 2, requests) +} + type roundTripperFunc func(*http.Request) (*http.Response, error) func (fn roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {