seafile,pcloud,pikpak,zoho,imagekit: stop multipart writers after failed uploads - fixes #10168

Co-authored-by: Florian Josef Altendorfer <341419330+florianaltendorfer@users.noreply.github.com>

backend/imagekit/client/upload.go

@@ -77,6 +77,8 @@ func (ik *ImageKit) Upload(ctx context.Context, file io.Reader, param UploadPara
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to make multipart upload: %w", err)
 	}
+	// Stop the form writer reading file if the request fails before sending it all
+	defer func() { _ = formReader.Close() }()
 
 	opts := rest.Opts{
 		Method:      "POST",

backend/pcloud/pcloud.go

@@ -1537,6 +1537,8 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		if err != nil {
 			return fmt.Errorf("failed to make multipart upload for 0 length file: %w", err)
 		}
+		// Stop the form writer reading in if the request fails before sending it all
+		defer func() { _ = formReader.Close() }()
 
 		contentLength := overhead + size
 

backend/pikpak/pikpak.go

@@ -1388,6 +1388,8 @@ func (f *Fs) uploadByForm(ctx context.Context, in io.Reader, name string, size i
 	if err != nil {
 		return fmt.Errorf("failed to make multipart upload: %w", err)
 	}
+	// Stop the form writer reading in if the request fails before sending it all
+	defer func() { _ = formReader.Close() }()
 
 	contentLength := overhead + size
 	opts := rest.Opts{

backend/seafile/seafile_internal_test.go

@@ -2,12 +2,20 @@ package seafile
 
 import (
 	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"path"
 	"testing"
+	"time"
 
 	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fs/asyncreader"
 	"github.com/rclone/rclone/fs/config/configmap"
 	"github.com/rclone/rclone/fs/config/obscure"
+	"github.com/rclone/rclone/lib/pacer"
+	"github.com/rclone/rclone/lib/readers"
+	"github.com/rclone/rclone/lib/rest"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/text/unicode/norm"
@@ -249,3 +257,35 @@ func Test2FAStateMachine(t *testing.T) {
 		})
 	}
 }
+
+// Test that a failed upload stops reading the source so it can be closed
+func TestUploadErrorReleasesSource(t *testing.T) {
+	ctx := context.Background()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Reply without reading the body, like a server rejecting the upload early
+		http.Error(w, "rejected", http.StatusBadRequest)
+	}))
+	defer server.Close()
+
+	f := &Fs{
+		srv:   rest.NewClient(server.Client()),
+		pacer: fs.NewPacer(ctx, pacer.NewDefault()),
+	}
+	// An AsyncReader holds its lock while writing to the upload
+	in, err := asyncreader.New(ctx, io.NopCloser(readers.NewPatternReader(64<<20)), 4)
+	require.NoError(t, err)
+
+	_, err = f.upload(ctx, in, server.URL+"/upload", "/file.bin")
+	require.Error(t, err)
+
+	closed := make(chan struct{})
+	go func() {
+		_ = in.Close()
+		close(closed)
+	}()
+	select {
+	case <-closed:
+	case <-time.After(10 * time.Second):
+		t.Fatal("source still in use after upload returned")
+	}
+}

backend/seafile/webapi.go

@@ -693,6 +693,8 @@ func (f *Fs) upload(ctx context.Context, in io.Reader, uploadLink, filePath stri
 	if err != nil {
 		return nil, fmt.Errorf("failed to make multipart upload: %w", err)
 	}
+	// Stop the form writer reading in if the request fails before sending it all
+	defer func() { _ = formReader.Close() }()
 
 	opts := rest.Opts{
 		Method:      "POST",

backend/zoho/zoho.go

@@ -1277,6 +1277,8 @@ func (f *Fs) upload(ctx context.Context, name string, parent string, size int64,
 	if err != nil {
 		return nil, fmt.Errorf("failed to make multipart upload: %w", err)
 	}
+	// Stop the form writer reading in if the request fails before sending it all
+	defer func() { _ = formReader.Close() }()
 
 	contentLength := overhead + size
 	opts := rest.Opts{