operations: fix ignored error in rcat probe reads

Return the first non-EOF input error encountered while Rcat probes whether an
upload is small. Previously that error was ignored and the full probe buffer,
including bytes that were never read, could be passed to Put or PutStream.

fs/operations/operations.go

@@ -1433,6 +1433,8 @@ func rcatSrc(ctx context.Context, fdst fs.Fs, dstFileName string, in io.ReadClos
 	if n, err := io.ReadFull(trackingIn, buf); err == io.EOF || err == io.ErrUnexpectedEOF {
 		fileIsSmall = true
 		buf = buf[:n]
+	} else if err != nil {
+		return nil, fmt.Errorf("failed to read upload input: %w", err)
 	}
 
 	// Read the data we have already read in buf and any further unread

fs/operations/operations_test.go

@@ -2168,3 +2168,32 @@ func TestRemoveExisting(t *testing.T) {
 	cleanup(&returnedError)
 	r.CheckRemoteItems(t)
 }
+
+func TestRcatInputFailurePreservesDestination(t *testing.T) {
+	ctx := context.Background()
+	root := t.TempDir()
+	require.NoError(t, os.WriteFile(root+"/target", []byte("original"), 0600))
+	f, err := fs.NewFs(ctx, root)
+	require.NoError(t, err)
+	calls := 0
+	original := f.Features().PutStream
+	f.Features().PutStream = func(ctx context.Context, in io.Reader, src fs.ObjectInfo, opts ...fs.OpenOption) (fs.Object, error) {
+		calls++
+		return original(ctx, in, src, opts...)
+	}
+	inputErr := errors.New("source interrupted")
+	for _, name := range []string{"target", "missing"} {
+		_, err = operations.Rcat(ctx, f, name, io.NopCloser(io.MultiReader(strings.NewReader("prefix"), rcatFailedInput{inputErr})), time.Now(), nil)
+		require.ErrorIs(t, err, inputErr)
+		require.Zero(t, calls)
+		b, readErr := os.ReadFile(root + "/target")
+		require.NoError(t, readErr)
+		require.Equal(t, "original", string(b))
+		_, statErr := os.Stat(root + "/missing")
+		require.True(t, os.IsNotExist(statErr))
+	}
+}
+
+type rcatFailedInput struct{ err error }
+
+func (r rcatFailedInput) Read([]byte) (int, error) { return 0, r.err }