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

@@ -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
 }

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) {