stores: fix completing very large mutipart uploads (#651)

stores/multipart.go

@@ -372,10 +372,6 @@ func (s *SQLStore) CompleteMultipartUpload(ctx context.Context, bucket, path str
 		sum := h.Sum()
 		eTag = hex.EncodeToString(sum[:])
 
-		// Sort their primary keys to make sure retrieving them later will
-		// respect the part order.
-		sort.Sort(sortedSlices(slices))
-
 		// Create the object.
 		obj := dbObject{
 			DBBucketID: mu.DBBucketID,
@@ -389,14 +385,17 @@ func (s *SQLStore) CompleteMultipartUpload(ctx context.Context, bucket, path str
 			return fmt.Errorf("failed to create object: %w", err)
 		}
 
-		// Assign the right object id and unassign the multipart upload.
+		// Assign the right object id and unassign the multipart upload.  Also
+		// clear the ID to make sure new slices are created with IDs in
+		// ascending order.
 		for i := range slices {
+			slices[i].ID = 0
 			slices[i].DBObjectID = &obj.ID
 			slices[i].DBMultipartPartID = nil
 		}
 
 		// Save updated slices.
-		if err := tx.Save(slices).Error; err != nil {
+		if err := tx.CreateInBatches(slices, 100).Error; err != nil {
 			return fmt.Errorf("failed to save slices: %w", err)
 		}
 
@@ -414,20 +413,6 @@ func (s *SQLStore) CompleteMultipartUpload(ctx context.Context, bucket, path str
 	}, nil
 }
 
-type sortedSlices []dbSlice
-
-func (s sortedSlices) Len() int {
-	return len(s)
-}
-
-func (s sortedSlices) Less(i, j int) bool {
-	return s[i].ID < s[j].ID
-}
-
-func (s sortedSlices) Swap(i, j int) {
-	s[i].ID, s[j].ID = s[j].ID, s[i].ID
-}
-
 func (u dbMultipartUpload) convert() (api.MultipartUpload, error) {
 	var key object.EncryptionKey
 	if err := key.UnmarshalText(u.Key); err != nil {

stores/multipart_test.go

@@ -70,9 +70,20 @@ func TestMultipartUploadWithUploadPackingRegression(t *testing.T) {
 			ETag:       etag,
 		})
 	}
-	_, err = db.CompleteMultipartUpload(ctx, api.DefaultBucketName, objName, resp.UploadID, parts)
-	if err != nil {
+
+	// Complete the upload. Check that the number of slices stays the same.
+	var nSlicesBefore int64
+	var nSlicesAfter int64
+	if err := db.db.Model(&dbSlice{}).Count(&nSlicesBefore).Error; err != nil {
+		t.Fatal(err)
+	} else if nSlicesBefore == 0 {
+		t.Fatal("expected some slices")
+	} else if _, err = db.CompleteMultipartUpload(ctx, api.DefaultBucketName, objName, resp.UploadID, parts); err != nil {
+		t.Fatal(err)
+	} else if err := db.db.Model(&dbSlice{}).Count(&nSlicesAfter).Error; err != nil {
 		t.Fatal(err)
+	} else if nSlicesBefore != nSlicesAfter {
+		t.Fatalf("expected number of slices to stay the same, but got %v before and %v after", nSlicesBefore, nSlicesAfter)
 	}
 
 	// Fetch the object.