fix(teldrive): single upload wrong chunk allocate

Author: TwoOnefour

drivers/teldrive/driver.go

@@ -183,11 +183,7 @@ func (d *Teldrive) Put(ctx context.Context, dstDir model.Obj, file model.FileStr
 		return d.touch(file.GetName(), dstDir.GetPath())
 	}
 
-	if totalParts <= 1 {
-		return d.doSingleUpload(ctx, dstDir, file, up, totalParts, chunkSize, fileId)
-	}
-
-	return d.doMultiUpload(ctx, dstDir, file, up, maxRetried, totalParts, chunkSize, fileId)
+	return d.doSingleUpload(ctx, dstDir, file, up, maxRetried, totalParts, chunkSize, fileId)
 }
 
 func (d *Teldrive) GetArchiveMeta(ctx context.Context, obj model.Obj, args model.ArchiveArgs) (model.ArchiveMeta, error) {

drivers/teldrive/upload.go

@@ -4,9 +4,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"sort"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/OpenListTeam/OpenList/v4/drivers/base"
@@ -16,10 +14,7 @@ import (
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 	"github.com/avast/retry-go"
 	"github.com/go-resty/resty/v2"
-	"github.com/pkg/errors"
 	"golang.org/x/net/context"
-	"golang.org/x/sync/errgroup"
-	"golang.org/x/sync/semaphore"
 )
 
 // create empty file
@@ -101,12 +96,10 @@ func (d *Teldrive) getFilePart(fileId string) ([]FilePart, error) {
 	return uploadedParts, nil
 }
 
-func (d *Teldrive) singleUploadRequest(fileId string, callback base.ReqCallback, resp interface{}) error {
+func (d *Teldrive) singleUploadRequest(ctx context.Context, fileId string, callback base.ReqCallback, resp interface{}) error {
 	url := d.Address + "/api/uploads/" + fileId
 	client := resty.New().SetTimeout(0)
 
-	ctx := context.Background()
-
 	req := client.R().
 		SetContext(ctx)
 	req.SetHeader("Cookie", d.Cookie)
@@ -135,16 +128,17 @@ func (d *Teldrive) singleUploadRequest(fileId string, callback base.ReqCallback,
 }
 
 func (d *Teldrive) doSingleUpload(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up model.UpdateProgress,
-	totalParts int, chunkSize int64, fileId string) error {
+	maxRetried, totalParts int, chunkSize int64, fileId string) error {
 
 	totalSize := file.GetSize()
 	var fileParts []FilePart
 	var uploaded int64 = 0
-	ss, err := stream.NewStreamSectionReader(file, int(totalSize), &up)
+	chunkSize = min(totalSize, chunkSize)
+	ss, err := stream.NewStreamSectionReader(file, int(chunkSize), &up)
 	if err != nil {
 		return err
 	}
-
+	chunkCnt := 0
 	for uploaded < totalSize {
 		if utils.IsCanceled(ctx) {
 			return ctx.Err()
@@ -155,19 +149,18 @@ func (d *Teldrive) doSingleUpload(ctx context.Context, dstDir model.Obj, file mo
 			return err
 		}
 		filePart := &FilePart{}
+		chunkCnt += 1
 		if err := retry.Do(func() error {
-
 			if _, err := rd.Seek(0, io.SeekStart); err != nil {
 				return err
 			}
-
-			if err := d.singleUploadRequest(fileId, func(req *resty.Request) {
+			if err := d.singleUploadRequest(ctx, fileId, func(req *resty.Request) {
 				uploadParams := map[string]string{
 					"partName": func() string {
-						digits := len(strconv.Itoa(totalParts))
-						return file.GetName() + fmt.Sprintf(".%0*d", digits, 1)
+						digits := len(strconv.Itoa(chunkCnt))
+						return file.GetName() + fmt.Sprintf(".%0*d", digits, chunkCnt)
 					}(),
-					"partNo":   strconv.Itoa(1),
+					"partNo":   strconv.Itoa(chunkCnt),
 					"fileName": file.GetName(),
 				}
 				req.SetQueryParams(uploadParams)
@@ -176,199 +169,24 @@ func (d *Teldrive) doSingleUpload(ctx context.Context, dstDir model.Obj, file mo
 			}, filePart); err != nil {
 				return err
 			}
-
 			return nil
 		},
 			retry.Context(ctx),
-			retry.Attempts(3),
+			retry.Attempts(uint(maxRetried)),
 			retry.DelayType(retry.BackOffDelay),
 			retry.Delay(time.Second)); err != nil {
 			return err
 		}
-
 		if filePart.Name != "" {
 			fileParts = append(fileParts, *filePart)
 			uploaded += curChunkSize
-			up(float64(uploaded) / float64(totalSize))
+			up(float64(uploaded) / float64(totalSize) * 100)
 			ss.FreeSectionReader(rd)
+		} else {
+			// For common situation this code won't reach
+			return fmt.Errorf("[Teldrive] upload chunk %d failed: filePart Somehow missing", chunkCnt)
 		}
-
 	}
 
 	return d.createFileOnUploadSuccess(file.GetName(), fileId, dstDir.GetPath(), fileParts, totalSize)
 }
-
-func (d *Teldrive) doMultiUpload(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up model.UpdateProgress,
-	maxRetried, totalParts int, chunkSize int64, fileId string) error {
-
-	concurrent := d.UploadConcurrency
-	g, ctx := errgroup.WithContext(ctx)
-	sem := semaphore.NewWeighted(int64(concurrent))
-	chunkChan := make(chan chunkTask, concurrent*2)
-	resultChan := make(chan FilePart, concurrent)
-	totalSize := file.GetSize()
-
-	ss, err := stream.NewStreamSectionReader(file, int(totalSize), &up)
-	if err != nil {
-		return err
-	}
-	ssLock := sync.Mutex{}
-	g.Go(func() error {
-		defer close(chunkChan)
-
-		chunkIdx := 0
-		for chunkIdx < totalParts {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			default:
-			}
-
-			offset := int64(chunkIdx) * chunkSize
-			curChunkSize := min(totalSize-offset, chunkSize)
-
-			ssLock.Lock()
-			reader, err := ss.GetSectionReader(offset, curChunkSize)
-			ssLock.Unlock()
-
-			if err != nil {
-				return err
-			}
-			task := chunkTask{
-				chunkIdx:  chunkIdx + 1,
-				chunkSize: curChunkSize,
-				fileName:  file.GetName(),
-				reader:    reader,
-				ss:        ss,
-			}
-			// freeSectionReader will be called in d.uploadSingleChunk
-			select {
-			case chunkChan <- task:
-				chunkIdx++
-			case <-ctx.Done():
-				return ctx.Err()
-			}
-		}
-		return nil
-	})
-	for i := 0; i < int(concurrent); i++ {
-		g.Go(func() error {
-			for task := range chunkChan {
-				if err := sem.Acquire(ctx, 1); err != nil {
-					return err
-				}
-
-				filePart, err := d.uploadSingleChunk(ctx, fileId, task, totalParts, maxRetried)
-				sem.Release(1)
-
-				if err != nil {
-					return fmt.Errorf("upload chunk %d failed: %w", task.chunkIdx, err)
-				}
-
-				select {
-				case resultChan <- *filePart:
-				case <-ctx.Done():
-					return ctx.Err()
-				}
-			}
-			return nil
-		})
-	}
-	var fileParts []FilePart
-	var collectErr error
-	collectDone := make(chan struct{})
-
-	go func() {
-		defer close(collectDone)
-		fileParts = make([]FilePart, 0, totalParts)
-
-		done := make(chan error, 1)
-		go func() {
-			done <- g.Wait()
-			close(resultChan)
-		}()
-
-		for {
-			select {
-			case filePart, ok := <-resultChan:
-				if !ok {
-					collectErr = <-done
-					return
-				}
-				fileParts = append(fileParts, filePart)
-			case err := <-done:
-				collectErr = err
-				return
-			}
-		}
-	}()
-
-	<-collectDone
-
-	if collectErr != nil {
-		return fmt.Errorf("multi-upload failed: %w", collectErr)
-	}
-	sort.Slice(fileParts, func(i, j int) bool {
-		return fileParts[i].PartNo < fileParts[j].PartNo
-	})
-
-	return d.createFileOnUploadSuccess(file.GetName(), fileId, dstDir.GetPath(), fileParts, totalSize)
-}
-
-func (d *Teldrive) uploadSingleChunk(ctx context.Context, fileId string, task chunkTask, totalParts, maxRetried int) (*FilePart, error) {
-	filePart := &FilePart{}
-	retryCount := 0
-	defer task.ss.FreeSectionReader(task.reader)
-
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		default:
-		}
-
-		if existingPart, err := d.checkFilePartExist(fileId, task.chunkIdx); err == nil && existingPart.Name != "" {
-			return &existingPart, nil
-		}
-
-		err := d.singleUploadRequest(fileId, func(req *resty.Request) {
-			uploadParams := map[string]string{
-				"partName": func() string {
-					digits := len(strconv.Itoa(totalParts))
-					return task.fileName + fmt.Sprintf(".%0*d", digits, task.chunkIdx)
-				}(),
-				"partNo":   strconv.Itoa(task.chunkIdx),
-				"fileName": task.fileName,
-			}
-			req.SetQueryParams(uploadParams)
-			req.SetBody(driver.NewLimitedUploadStream(ctx, task.reader))
-			req.SetHeader("Content-Length", strconv.Itoa(int(task.chunkSize)))
-		}, filePart)
-
-		if err == nil {
-			return filePart, nil
-		}
-
-		if retryCount >= maxRetried {
-			return nil, fmt.Errorf("upload failed after %d retries: %w", maxRetried, err)
-		}
-
-		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
-			continue
-		}
-
-		retryCount++
-		utils.Log.Errorf("[Teldrive] upload error: %v, retrying %d times", err, retryCount)
-
-		backoffDuration := time.Duration(retryCount*retryCount) * time.Second
-		if backoffDuration > 30*time.Second {
-			backoffDuration = 30 * time.Second
-		}
-
-		select {
-		case <-time.After(backoffDuration):
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		}
-	}
-}