drive: reuse resumable-upload chunk buffers via multipart.NewRW

Each resumable upload allocated a fresh chunk-sized buffer (8 MiB by
default), so bulk transfers of many files churned allocations and GC.
Buffer chunks with multipart.NewRW instead — the global page pool used
by the other backends — so chunk memory is reused across uploads and
bounded by rclone's central memory management. The pool.RW is seekable,
which also keeps chunk reads repeatable for retries.

The pool.RW implements io.Closer, so http.NewRequestWithContext upgraded
it to the request body and the transport closed it after each attempt,
returning its pages to the global pool — a chunk retried after a 5xx
then read a freed buffer and panicked in pool.(*RW).readPage. Wrap the
request body in readers.NoCloser so the transport can't take ownership
and the upload loop remains solely responsible for the buffer's
lifetime. Add a regression test that fails a chunk with a 500 and then
accepts the retry; it reproduces the panic without the fix.

Fixes #9684
This commit is contained in:
SillyZir
2026-08-30 13:07:41 +01:00
committed by Nick Craig-Wood
parent 4722b94d1a
commit 393544b116
2 changed files with 87 additions and 18 deletions
+30 -18
View File
@@ -11,7 +11,6 @@
package drive package drive
import ( import (
"bytes"
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
@@ -22,6 +21,7 @@ import (
"github.com/rclone/rclone/fs" "github.com/rclone/rclone/fs"
"github.com/rclone/rclone/fs/fserrors" "github.com/rclone/rclone/fs/fserrors"
"github.com/rclone/rclone/lib/multipart"
"github.com/rclone/rclone/lib/readers" "github.com/rclone/rclone/lib/readers"
"google.golang.org/api/drive/v3" "google.golang.org/api/drive/v3"
"google.golang.org/api/googleapi" "google.golang.org/api/googleapi"
@@ -112,8 +112,13 @@ func (f *Fs) Upload(ctx context.Context, in io.Reader, size int64, contentType,
} }
// Make an http.Request for the range passed in // Make an http.Request for the range passed in
//
// The body is wrapped in readers.NoCloser so the transport can't upgrade it
// to an io.Closer and close it after each attempt — the chunk buffer is
// pool-backed, so an early close would free its pages and a retry would then
// read a dead buffer. The upload loop owns the buffer's lifetime.
func (rx *resumableUpload) makeRequest(ctx context.Context, start int64, body io.ReadSeeker, reqSize int64) *http.Request { func (rx *resumableUpload) makeRequest(ctx context.Context, start int64, body io.ReadSeeker, reqSize int64) *http.Request {
req, _ := http.NewRequestWithContext(ctx, "POST", rx.URI, body) req, _ := http.NewRequestWithContext(ctx, "POST", rx.URI, readers.NoCloser(body))
req.ContentLength = reqSize req.ContentLength = reqSize
totalSize := "*" totalSize := "*"
if rx.ContentLength >= 0 { if rx.ContentLength >= 0 {
@@ -168,37 +173,40 @@ func (rx *resumableUpload) Upload(ctx context.Context) (*drive.File, error) {
start := int64(0) start := int64(0)
var StatusCode int var StatusCode int
var err error var err error
buf := make([]byte, int(rx.f.opt.ChunkSize)) chunkSize := int64(rx.f.opt.ChunkSize)
for finished := false; !finished; { for finished := false; !finished; {
var reqSize int64 reqSize := chunkSize
var chunk io.ReadSeeker
if rx.ContentLength >= 0 { if rx.ContentLength >= 0 {
// If size known use repeatable reader for smoother bwlimit
if start >= rx.ContentLength { if start >= rx.ContentLength {
break break
} }
reqSize = min(rx.ContentLength-start, int64(rx.f.opt.ChunkSize)) reqSize = min(rx.ContentLength-start, chunkSize)
chunk = readers.NewRepeatableLimitReaderBuffer(rx.Media, buf, reqSize) }
} else {
// If size unknown read into buffer // Buffer the chunk in memory from the global pool so reads are
var n int // repeatable for retries
n, err = readers.ReadFill(rx.Media, buf) rw := multipart.NewRW()
var n int64
n, err = io.CopyN(rw, rx.Media, reqSize)
if rx.ContentLength < 0 {
if err == io.EOF { if err == io.EOF {
// Send the last chunk with the correct ContentLength // Send the last chunk with the correct ContentLength
// otherwise Google doesn't know we've finished // otherwise Google doesn't know we've finished
rx.ContentLength = start + int64(n) rx.ContentLength = start + n
finished = true finished = true
} else if err != nil { err = nil
return nil, err
} }
reqSize = int64(n) reqSize = n
chunk = bytes.NewReader(buf[:reqSize]) }
if err != nil {
_ = rw.Close()
return nil, err
} }
// Transfer the chunk // Transfer the chunk
err = rx.f.pacer.Call(func() (bool, error) { err = rx.f.pacer.Call(func() (bool, error) {
fs.Debugf(rx.remote, "Sending chunk %d length %d", start, reqSize) fs.Debugf(rx.remote, "Sending chunk %d length %d", start, reqSize)
StatusCode, err = rx.transferChunk(ctx, start, chunk, reqSize) StatusCode, err = rx.transferChunk(ctx, start, rw, reqSize)
again, err := rx.f.shouldRetry(ctx, err) again, err := rx.f.shouldRetry(ctx, err)
if StatusCode == statusResumeIncomplete || StatusCode == http.StatusCreated || StatusCode == http.StatusOK { if StatusCode == statusResumeIncomplete || StatusCode == http.StatusCreated || StatusCode == http.StatusOK {
again = false again = false
@@ -206,9 +214,13 @@ func (rx *resumableUpload) Upload(ctx context.Context) (*drive.File, error) {
} }
return again, err return again, err
}) })
closeErr := rw.Close()
if err != nil { if err != nil {
return nil, err return nil, err
} }
if closeErr != nil {
return nil, closeErr
}
start += reqSize start += reqSize
} }
+57
View File
@@ -0,0 +1,57 @@
package drive
import (
"bytes"
"context"
"io"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"github.com/rclone/rclone/fs"
"github.com/rclone/rclone/lib/pacer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestResumableUploadRetry checks that a chunk which fails with a 5xx is
// retried successfully. The chunk buffer is pool-backed and implements
// io.Closer, so if it reaches the http transport unwrapped the transport
// closes it after the failed attempt, returning its pages to the pool, and
// the retry then reads a freed buffer.
func TestResumableUploadRetry(t *testing.T) {
content := bytes.Repeat([]byte("resumable"), 512)
var attempts atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(r.Body)
require.NoError(t, err)
if attempts.Add(1) == 1 {
w.WriteHeader(http.StatusInternalServerError)
return
}
assert.Equal(t, content, body, "retried chunk should re-send the full chunk")
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"id":"fake-id","name":"remote"}`))
}))
defer server.Close()
f := &Fs{
pacer: fs.NewPacer(context.Background(), pacer.NewGoogleDrive()),
client: server.Client(),
}
f.opt.ChunkSize = fs.SizeSuffix(len(content))
rx := &resumableUpload{
f: f,
remote: "remote",
URI: server.URL,
Media: bytes.NewReader(content),
MediaType: "application/octet-stream",
ContentLength: int64(len(content)),
}
info, err := rx.Upload(context.Background())
require.NoError(t, err)
require.NotNil(t, info)
assert.Equal(t, "fake-id", info.Id)
assert.Equal(t, int32(2), attempts.Load(), "expected exactly one failed attempt and one retry")
}