sync: fix temporary name when create a file (#4215)

juicedata · Feb 3, 2024 · e5603e2 · e5603e2
1 parent 64bb330
commit e5603e2
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 13 deletions.
diff --git a/pkg/object/file.go b/pkg/object/file.go
@@ -132,14 +132,23 @@ func (d *filestore) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	return f, nil
 }
 
-func (d *filestore) Put(key string, in io.Reader) error {
+func (d *filestore) Put(key string, in io.Reader) (err error) {
 	p := d.path(key)
 
 	if strings.HasSuffix(key, dirSuffix) || key == "" && strings.HasSuffix(d.root, dirSuffix) {
 		return os.MkdirAll(p, os.FileMode(0755))
 	}
 
-	tmp := filepath.Join(filepath.Dir(p), "."+filepath.Base(p)+".tmp"+strconv.Itoa(rand.Int()))
+	name := filepath.Base(p)
+	if len(name) > 200 {
+		name = name[:200]
+	}
+	tmp := filepath.Join(filepath.Dir(p), "."+name+".tmp"+strconv.Itoa(rand.Int()))
+	defer func() {
+		if err != nil {
+			_ = os.Remove(tmp)
+		}
+	}()
 	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
 	if err != nil && os.IsNotExist(err) {
 		if err := os.MkdirAll(filepath.Dir(p), os.FileMode(0755)); err != nil {
@@ -150,11 +159,6 @@ func (d *filestore) Put(key string, in io.Reader) error {
 	if err != nil {
 		return err
 	}
-	defer func() {
-		if err != nil {
-			_ = os.Remove(tmp)
-		}
-	}()
 
 	if TryCFR {
 		_, err = io.Copy(f, in)

diff --git a/pkg/object/filesystem_test.go b/pkg/object/filesystem_test.go
@@ -195,4 +195,10 @@ func testFileSystem(t *testing.T, s ObjectStorage) {
 			t.Fatalf("size of target(file) should be 10")
 		}
 	}
+
+	// put a file with very long name
+	longName := strings.Repeat("a", 255)
+	if err := s.Put("dir/"+longName, bytes.NewReader([]byte{0})); err != nil {
+		t.Fatalf("PUT a file with long name `%s` failed: %q", longName, err)
+	}
 }
diff --git a/pkg/object/hdfs.go b/pkg/object/hdfs.go
@@ -114,14 +114,22 @@ func (h *hdfsclient) Get(key string, off, limit int64) (io.ReadCloser, error) {
 
 const abcException = "org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException"
 
-func (h *hdfsclient) Put(key string, in io.Reader) error {
+func (h *hdfsclient) Put(key string, in io.Reader) (err error) {
 	path := h.path(key)
 	if strings.HasSuffix(path, dirSuffix) {
 		return h.c.MkdirAll(path, os.FileMode(0755))
 	}
-	tmp := filepath.Join(filepath.Dir(path), fmt.Sprintf(".%s.tmp.%d", filepath.Base(path), rand.Int()))
+	name := filepath.Base(path)
+	if len(name) > 200 {
+		name = name[:200]
+	}
+	tmp := filepath.Join(filepath.Dir(path), fmt.Sprintf(".%s.tmp.%d", name, rand.Int()))
+	defer func() {
+		if err != nil {
+			_ = h.c.Remove(tmp)
+		}
+	}()
 	f, err := h.c.CreateFile(tmp, 3, 128<<20, 0755)
-	defer func() { _ = h.c.Remove(tmp) }()
 	if err != nil {
 		if pe, ok := err.(*os.PathError); ok && pe.Err == os.ErrNotExist {
 			_ = h.c.MkdirAll(filepath.Dir(path), 0755)

diff --git a/pkg/object/sftp.go b/pkg/object/sftp.go
@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math/rand"
 	"net"
 	"net/url"
 	"os"
@@ -202,7 +203,7 @@ func (f *sftpStore) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	return ff, err
 }
 
-func (f *sftpStore) Put(key string, in io.Reader) error {
+func (f *sftpStore) Put(key string, in io.Reader) (err error) {
 	c, err := f.getSftpConnection()
 	if err != nil {
 		return err
@@ -216,7 +217,16 @@ func (f *sftpStore) Put(key string, in io.Reader) error {
 	if err := c.sftpClient.MkdirAll(filepath.Dir(p)); err != nil {
 		return err
 	}
-	tmp := filepath.Join(filepath.Dir(p), "."+filepath.Base(p)+".tmp")
+	name := filepath.Base(p)
+	if len(name) > 200 {
+		name = name[:200]
+	}
+	tmp := filepath.Join(filepath.Dir(p), fmt.Sprintf(".%s.tmp.%d", name, rand.Int()))
+	defer func() {
+		if err != nil {
+			_ = c.sftpClient.Remove(tmp)
+		}
+	}()
 	if runtime.GOOS == "windows" {
 		tmp = strings.Replace(tmp, "\\", "/", -1)
 	}
@@ -225,7 +235,6 @@ func (f *sftpStore) Put(key string, in io.Reader) error {
 	if err != nil {
 		return err
 	}
-	defer func() { _ = c.sftpClient.Remove(tmp) }()
 	buf := bufPool.Get().(*[]byte)
 	defer bufPool.Put(buf)
 	_, err = io.CopyBuffer(ff, in, *buf)