shell: preserve completed uploads for bucket-root objects (#12545)

Signed-off-by: rleungx <rleungx@gmail.com>

weed/shell/command_s3_clean_uploads.go

@@ -137,7 +137,7 @@ func (c *commandS3CleanUploads) uploadCompleted(filerClient filer_pb.FilerClient
 	if dir == "." {
 		dir = ""
 	}
-	objectDir := util.FullPath(bucketDir + "/" + dir)
+	objectDir := util.FullPath(path.Join(bucketDir, dir))
 
 	completed := false
 	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {

weed/shell/command_s3_clean_uploads_test.go

@@ -29,6 +29,7 @@ type stubFilerSvc struct {
 	lookupResp *filer_pb.LookupDirectoryEntryResponse
 	lookupErr  error
 	listResp   []*filer_pb.Entry
+	listDir    string
 	listErr    error
 	lookupReq  *filer_pb.LookupDirectoryEntryRequest
 	listReq    *filer_pb.ListEntriesRequest
@@ -44,6 +45,9 @@ func (s *stubFilerSvc) ListEntries(ctx context.Context, req *filer_pb.ListEntrie
 	if s.listErr != nil {
 		return nil, s.listErr
 	}
+	if s.listDir != "" && req.Directory != s.listDir {
+		return &stubListStream{}, nil
+	}
 	return &stubListStream{entries: s.listResp}, nil
 }
 
@@ -81,6 +85,34 @@ func completedEntry(uploadId string) *filer_pb.Entry {
 	}
 }
 
+func TestUploadCompletedVersionPaths(t *testing.T) {
+	for _, tc := range []struct {
+		key       string
+		objectDir string
+		name      string
+	}{
+		{key: "obj", objectDir: "/buckets/b", name: "obj"},
+		{key: "dir/obj", objectDir: "/buckets/b/dir", name: "obj"},
+		{key: "dir/", objectDir: "/buckets/b/dir", name: "dir"},
+	} {
+		t.Run(tc.key, func(t *testing.T) {
+			svc := &stubFilerSvc{
+				lookupErr: filer_pb.ErrNotFound,
+				listDir:   tc.objectDir + "/" + tc.name + s3_constants.VersionsFolder,
+				listResp:  []*filer_pb.Entry{completedEntry("up1")},
+			}
+			c := &commandS3CleanUploads{}
+			completed, err := c.uploadCompleted(stubFilerClient{client: svc}, "/buckets/b", uploadEntry("up1", tc.key))
+			if err != nil || !completed {
+				t.Fatalf("got (%v, %v), want (true, nil)", completed, err)
+			}
+			if svc.lookupReq.Directory != tc.objectDir || svc.lookupReq.Name != tc.name {
+				t.Fatalf("lookup = %s/%s, want %s/%s", svc.lookupReq.Directory, svc.lookupReq.Name, tc.objectDir, tc.name)
+			}
+		})
+	}
+}
+
 func TestUploadCompleted(t *testing.T) {
 	c := &commandS3CleanUploads{}