diff --git a/backend/box/upload.go b/backend/box/upload.go index a9b3a9f9e..b79af2394 100644 --- a/backend/box/upload.go +++ b/backend/box/upload.go @@ -3,7 +3,6 @@ package box import ( - "bytes" "context" "crypto/sha1" "encoding/base64" @@ -20,6 +19,7 @@ import ( "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fs/accounting" "github.com/rclone/rclone/lib/atexit" + "github.com/rclone/rclone/lib/multipart" "github.com/rclone/rclone/lib/rest" ) @@ -55,9 +55,15 @@ func sha1Digest(digest []byte) string { } // uploadPart uploads a part in an upload session -func (o *Object) uploadPart(ctx context.Context, SessionID string, offset, totalSize int64, chunk []byte, wrap accounting.WrapFn, options ...fs.OpenOption) (response *api.UploadPartResponse, err error) { - chunkSize := int64(len(chunk)) - sha1sum := sha1.Sum(chunk) +// +// chunk must contain exactly the part's data and is read from the start +// on each attempt. +func (o *Object) uploadPart(ctx context.Context, SessionID string, offset, totalSize int64, chunk io.ReadSeeker, wrap accounting.WrapFn, options ...fs.OpenOption) (response *api.UploadPartResponse, err error) { + hasher := sha1.New() + chunkSize, err := io.Copy(hasher, chunk) + if err != nil { + return nil, err + } opts := rest.Opts{ Method: "PUT", Path: "/files/upload_sessions/" + SessionID, @@ -67,12 +73,16 @@ func (o *Object) uploadPart(ctx context.Context, SessionID string, offset, total ContentRange: fmt.Sprintf("bytes %d-%d/%d", offset, offset+chunkSize-1, totalSize), Options: options, ExtraHeaders: map[string]string{ - "Digest": sha1Digest(sha1sum[:]), + "Digest": sha1Digest(hasher.Sum(nil)), }, } var resp *http.Response err = o.fs.pacer.Call(func() (bool, error) { - opts.Body = wrap(bytes.NewReader(chunk)) + _, err = chunk.Seek(0, io.SeekStart) + if err != nil { + return false, err + } + opts.Body = wrap(chunk) resp, err = o.fs.srv.CallJSON(ctx, &opts, nil, &response) return shouldRetry(ctx, resp, err) }) @@ -213,27 +223,25 @@ outer: reqSize := min(remaining, chunkSize) - // Make a block of memory - buf := make([]byte, reqSize) - - // Read the chunk - _, err = io.ReadFull(in, buf) + // Read the chunk into memory from the global pool, hashing + // it for the whole file hash (must be done sequentially) + rw := multipart.NewRW().Reserve(reqSize) + _, err = io.CopyN(rw, io.TeeReader(in, hash), reqSize) if err != nil { + _ = rw.Close() err = fmt.Errorf("multipart upload failed to read source: %w", err) break outer } - // Make the global hash (must be done sequentially) - _, _ = hash.Write(buf) - // Transfer the chunk wg.Add(1) o.fs.uploadToken.Get() go func(part int, position int64) { defer wg.Done() defer o.fs.uploadToken.Put() + defer func() { _ = rw.Close() }() fs.Debugf(o, "Uploading part %d/%d offset %v/%v part size %v", part+1, session.TotalParts, fs.SizeSuffix(position), fs.SizeSuffix(size), fs.SizeSuffix(chunkSize)) - partResponse, err := o.uploadPart(ctx, session.ID, position, size, buf, wrap, options...) + partResponse, err := o.uploadPart(ctx, session.ID, position, size, rw, wrap, options...) if err != nil { err = fmt.Errorf("multipart upload failed to upload part: %w", err) select {