From 885e2478e7758633530ac1c07fdf6b7110b9edbe Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Sun, 30 Aug 2026 17:34:25 +0100 Subject: [PATCH] jottacloud: buffer MD5 pre-reads in the global memory pool When the source has no MD5, Update reads the whole file into memory to hash it before uploading if it is under --jottacloud-md5-memory-limit. This used io.ReadAll, which grows a fresh heap slice per file (up to 10 MiB by default, roughly doubled by the growth strategy), so syncs of many files churned allocations and GC. Buffer the data with multipart.NewRW instead so the memory comes from rclone's global pool, is reused across uploads and is released by the existing cleanup function. Unknown sized streams previously took the in-memory branch regardless of the limit, so an rcat of an arbitrarily large stream could read it all into memory. Spool those to the temporary file instead, as is already done for files over the limit. The buffered body is sent through lib/rest, which wraps request bodies in readers.NoCloser, so the transport can't close the pooled buffer early. --- backend/jottacloud/jottacloud.go | 20 ++++++----- .../jottacloud/jottacloud_internal_test.go | 35 ++++++++++++++++++- 2 files changed, 45 insertions(+), 10 deletions(-) diff --git a/backend/jottacloud/jottacloud.go b/backend/jottacloud/jottacloud.go index aad41a422..6af70b99b 100644 --- a/backend/jottacloud/jottacloud.go +++ b/backend/jottacloud/jottacloud.go @@ -34,6 +34,7 @@ import ( "github.com/rclone/rclone/fs/hash" "github.com/rclone/rclone/fs/list" "github.com/rclone/rclone/lib/encoder" + "github.com/rclone/rclone/lib/multipart" "github.com/rclone/rclone/lib/oauthutil" "github.com/rclone/rclone/lib/pacer" "github.com/rclone/rclone/lib/rest" @@ -1924,8 +1925,9 @@ func readMD5(in io.Reader, size, threshold int64) (md5sum string, out io.Reader, // nothing to clean up by default cleanup = func() {} - // don't cache small files on disk to reduce wear of the disk - if size > threshold { + // don't cache small files on disk to reduce wear of the disk, but + // spool unknown sized streams there as they can't be bounded in memory + if size > threshold || size < 0 { var tempFile *os.File // create the cache file @@ -1955,15 +1957,15 @@ func readMD5(in io.Reader, size, threshold int64) (md5sum string, out io.Reader, // replace the already read source with a reader of our cached file out = tempFile } else { - // that's a small file, just read it into memory - var inData []byte - inData, err = io.ReadAll(teeReader) - if err != nil { + // that's a small file, just read it into memory from the global pool + rw := multipart.NewRW() + cleanup = func() { + _ = rw.Close() + } + if _, err = io.Copy(rw, teeReader); err != nil { return } - - // set the reader to our read memory block - out = bytes.NewReader(inData) + out = rw } return hex.EncodeToString(md5Hasher.Sum(nil)), out, cleanup, nil } diff --git a/backend/jottacloud/jottacloud_internal_test.go b/backend/jottacloud/jottacloud_internal_test.go index db9d6a157..1889400c1 100644 --- a/backend/jottacloud/jottacloud_internal_test.go +++ b/backend/jottacloud/jottacloud_internal_test.go @@ -5,12 +5,14 @@ import ( "crypto/md5" "fmt" "io" + "os" "testing" "time" "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fstest" "github.com/rclone/rclone/fstest/fstests" + "github.com/rclone/rclone/lib/pool" "github.com/rclone/rclone/lib/random" "github.com/rclone/rclone/lib/readers" "github.com/stretchr/testify/assert" @@ -28,11 +30,16 @@ func TestReadMD5(t *testing.T) { wantMD5 := fmt.Sprintf("%x", hasher.Sum(nil)) for _, threshold := range []int64{512, 1024, 10 * 1024, 20 * 1024} { t.Run(fmt.Sprintf("%d", threshold), func(t *testing.T) { + inUse := pool.Global().InUse() in := readers.NewPatternReader(size) gotMD5, out, cleanup, err := readMD5(in, size, threshold) - defer cleanup() require.NoError(t, err) assert.Equal(t, wantMD5, gotMD5) + if size > threshold { + assert.IsType(t, (*os.File)(nil), out, "big files should be spooled to disk") + } else { + assert.IsType(t, (*pool.RW)(nil), out, "small files should be buffered in the pool") + } // check md5hash of out hasher := md5.New() @@ -41,10 +48,36 @@ func TestReadMD5(t *testing.T) { assert.Equal(t, n, size) outMD5 := fmt.Sprintf("%x", hasher.Sum(nil)) assert.Equal(t, wantMD5, outMD5) + + // check cleanup returns the buffer to the pool + cleanup() + assert.Equal(t, inUse, pool.Global().InUse(), "pool buffers leaked") }) } }) } + + // Unknown sized streams can't be bounded in memory so must be spooled to disk + t.Run("UnknownSize", func(t *testing.T) { + const size = 1024 + inUse := pool.Global().InUse() + hasher := md5.New() + _, err := io.Copy(hasher, readers.NewPatternReader(size)) + require.NoError(t, err) + wantMD5 := fmt.Sprintf("%x", hasher.Sum(nil)) + + gotMD5, out, cleanup, err := readMD5(readers.NewPatternReader(size), -1, 10*size) + require.NoError(t, err) + assert.Equal(t, wantMD5, gotMD5) + assert.IsType(t, (*os.File)(nil), out) + hasher.Reset() + n, err := io.Copy(hasher, out) + require.NoError(t, err) + assert.Equal(t, int64(size), n) + assert.Equal(t, wantMD5, fmt.Sprintf("%x", hasher.Sum(nil))) + cleanup() + assert.Equal(t, inUse, pool.Global().InUse()) + }) } func (f *Fs) InternalTestMetadata(t *testing.T) {