Feat: cancel upload session in slave node

pkg/filesystem/chunk/backoff/backoff.go

@@ -18,7 +18,7 @@ type ConstantBackoff struct {
 
 func (c *ConstantBackoff) Next() bool {
 	c.tried++
-	if c.tried >= c.Max {
+	if c.tried > c.Max {
 		return false
 	}
 

pkg/filesystem/chunk/chunk.go

@@ -49,7 +49,7 @@ func (c *ChunkGroup) Process(processor ChunkProcessFunc) error {
 	err := processor(c, io.LimitReader(c.file, int64(c.chunkSize)))
 	if err != nil {
 		if err != context.Canceled && c.file.Seekable() && c.backoff.Next() {
-			if _, seekErr := c.file.Seek(c.Start(), io.SeekStart); err != nil {
+			if _, seekErr := c.file.Seek(c.Start(), io.SeekStart); seekErr != nil {
 				return fmt.Errorf("failed to seek back to chunk start: %w, last error: %w", seekErr, err)
 			}
 

pkg/filesystem/driver/remote/client.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
 	"github.com/cloudreve/Cloudreve/v3/pkg/request"
 	"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
+	"github.com/cloudreve/Cloudreve/v3/pkg/util"
 	"github.com/gofrs/uuid"
 	"io"
 	"net/http"
@@ -26,7 +27,7 @@ const (
 	chunkRetrySleep = time.Duration(5) * time.Second
 )
 
-// Client to operate remote slave server
+// Client to operate uploading to remote slave server
 type Client interface {
 	// CreateUploadSession creates remote upload session
 	CreateUploadSession(ctx context.Context, session *serializer.UploadSession, ttl int64) error
@@ -34,6 +35,8 @@ type Client interface {
 	GetUploadURL(ttl int64, sessionID string) (string, string, error)
 	// Upload uploads file to remote server
 	Upload(ctx context.Context, file fsctx.FileHeader) error
+	// DeleteUploadSession deletes remote upload session
+	DeleteUploadSession(ctx context.Context, sessionID string) error
 }
 
 // NewClient creates new Client from given policy
@@ -84,7 +87,7 @@ func (c *remoteClient) Upload(ctx context.Context, file fsctx.FileHeader) error
 
 	overwrite := fileInfo.Mode&fsctx.Overwrite == fsctx.Overwrite
 
-	// Upload chunks
+	// Initial chunk groups
 	chunks := chunk.NewChunkGroup(file, c.policy.OptionsSerialized.ChunkSize, &backoff.ConstantBackoff{
 		Max:   model.GetIntSetting("onedrive_chunk_retries", 1),
 		Sleep: chunkRetrySleep,
@@ -94,9 +97,13 @@ func (c *remoteClient) Upload(ctx context.Context, file fsctx.FileHeader) error
 		return c.uploadChunk(ctx, session.Key, current.Index(), content, overwrite, current.Length())
 	}
 
+	// upload chunks
 	for chunks.Next() {
 		if err := chunks.Process(uploadFunc); err != nil {
-			// TODO 删除上传会话
+			if err := c.DeleteUploadSession(ctx, session.Key); err != nil {
+				util.Log().Warning("failed to delete upload session: %s", err)
+			}
+
 			return fmt.Errorf("failed to upload chunk #%d: %w", chunks.Index(), err)
 		}
 	}
@@ -104,6 +111,24 @@ func (c *remoteClient) Upload(ctx context.Context, file fsctx.FileHeader) error
 	return nil
 }
 
+func (c *remoteClient) DeleteUploadSession(ctx context.Context, sessionID string) error {
+	resp, err := c.httpClient.Request(
+		"DELETE",
+		"upload/"+sessionID,
+		nil,
+		request.WithContext(ctx),
+	).CheckHTTPResponse(200).DecodeResponse()
+	if err != nil {
+		return err
+	}
+
+	if resp.Code != 0 {
+		return serializer.NewErrorFromResponse(resp)
+	}
+
+	return nil
+}
+
 func (c *remoteClient) CreateUploadSession(ctx context.Context, session *serializer.UploadSession, ttl int64) error {
 	reqBodyEncoded, err := json.Marshal(map[string]interface{}{
 		"session": session,

pkg/filesystem/driver/remote/handler.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"net/http"
 	"net/url"
 	"path"
 	"strings"
@@ -26,10 +25,11 @@ type Driver struct {
 	Policy       *model.Policy
 	AuthInstance auth.Auth
 
-	client Client
+	uploadClient Client
 }
 
 // NewDriver initializes a new Driver from policy
+// TODO: refactor all method into upload client
 func NewDriver(policy *model.Policy) (*Driver, error) {
 	client, err := NewClient(policy)
 	if err != nil {
@@ -40,7 +40,7 @@ func NewDriver(policy *model.Policy) (*Driver, error) {
 		Policy:       policy,
 		Client:       request.NewClient(),
 		AuthInstance: auth.HMACAuth{[]byte(policy.SecretKey)},
-		client:       client,
+		uploadClient: client,
 	}, nil
 }
 
@@ -153,7 +153,7 @@ func (handler *Driver) Get(ctx context.Context, path string) (response.RSCloser,
 func (handler *Driver) Put(ctx context.Context, file fsctx.FileHeader) error {
 	defer file.Close()
 
-	return handler.client.Upload(ctx, file)
+	return handler.uploadClient.Upload(ctx, file)
 }
 
 // Delete 删除一个或多个文件，
@@ -281,12 +281,12 @@ func (handler *Driver) Token(ctx context.Context, ttl int64, uploadSession *seri
 
 	// 在从机端创建上传会话
 	uploadSession.Callback = apiURL.String()
-	if err := handler.client.CreateUploadSession(ctx, uploadSession, ttl); err != nil {
+	if err := handler.uploadClient.CreateUploadSession(ctx, uploadSession, ttl); err != nil {
 		return nil, err
 	}
 
 	// 获取上传地址
-	uploadURL, sign, err := handler.client.GetUploadURL(ttl, uploadSession.Key)
+	uploadURL, sign, err := handler.uploadClient.GetUploadURL(ttl, uploadSession.Key)
 	if err != nil {
 		return nil, fmt.Errorf("failed to sign upload url: %w", err)
 	}
@@ -299,30 +299,7 @@ func (handler *Driver) Token(ctx context.Context, ttl int64, uploadSession *seri
 	}, nil
 }
 
-func (handler *Driver) getUploadCredential(ctx context.Context, policy serializer.UploadPolicy, TTL int64) (serializer.UploadCredential, error) {
-	policyEncoded, err := policy.EncodeUploadPolicy()
-	if err != nil {
-		return serializer.UploadCredential{}, err
-	}
-
-	// 签名上传策略
-	uploadRequest, _ := http.NewRequest("POST", "/api/v3/slave/upload", nil)
-	uploadRequest.Header = map[string][]string{
-		"X-Cr-Policy":    {policyEncoded},
-		"X-Cr-Overwrite": {"false"},
-	}
-	auth.SignRequest(handler.AuthInstance, uploadRequest, TTL)
-
-	if credential, ok := uploadRequest.Header["Authorization"]; ok && len(credential) == 1 {
-		return serializer.UploadCredential{
-			Token:  credential[0],
-			Policy: policyEncoded,
-		}, nil
-	}
-	return serializer.UploadCredential{}, errors.New("无法签名上传策略")
-}
-
 // 取消上传凭证
 func (handler *Driver) CancelToken(ctx context.Context, uploadSession *serializer.UploadSession) error {
-	return nil
+	return handler.uploadClient.DeleteUploadSession(ctx, uploadSession.Key)
 }

routers/controllers/slave.go

@@ -28,77 +28,6 @@ func SlaveUpload(c *gin.Context) {
 	} else {
 		c.JSON(200, ErrorResponse(err))
 	}
-
-	//// 创建上下文
-	//ctx, cancel := context.WithCancel(context.Background())
-	//ctx = context.WithValue(ctx, fsctx.GinCtx, c)
-	//defer cancel()
-	//
-	//// 创建匿名文件系统
-	//fs, err := filesystem.NewAnonymousFileSystem()
-	//if err != nil {
-	//	c.JSON(200, serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err))
-	//	return
-	//}
-	//fs.Handler = local.Driver{}
-	//
-	//// 从请求中取得上传策略
-	//uploadPolicyRaw := c.GetHeader("X-Cr-Policy")
-	//if uploadPolicyRaw == "" {
-	//	c.JSON(200, serializer.ParamErr("未指定上传策略", nil))
-	//	return
-	//}
-	//
-	//// 解析上传策略
-	//uploadPolicy, err := serializer.DecodeUploadPolicy(uploadPolicyRaw)
-	//if err != nil {
-	//	c.JSON(200, serializer.ParamErr("上传策略格式有误", err))
-	//	return
-	//}
-	//ctx = context.WithValue(ctx, fsctx.UploadPolicyCtx, *uploadPolicy)
-	//
-	//// 取得文件大小
-	//fileSize, err := strconv.ParseUint(c.Request.Header.Get("Content-Length"), 10, 64)
-	//if err != nil {
-	//	c.JSON(200, ErrorResponse(err))
-	//	return
-	//}
-	//
-	//// 解码文件名和路径
-	//fileName, err := url.QueryUnescape(c.Request.Header.Get("X-Cr-FileName"))
-	//if err != nil {
-	//	c.JSON(200, ErrorResponse(err))
-	//	return
-	//}
-	//
-	//fileData := fsctx.FileStream{
-	//	MIMEType: c.Request.Header.Get("Content-Type"),
-	//	File:     c.Request.Body,
-	//	Name:     fileName,
-	//	Size:     fileSize,
-	//}
-	//
-	//// 给文件系统分配钩子
-	//fs.Use("BeforeUpload", filesystem.HookSlaveUploadValidate)
-	//fs.Use("AfterUploadCanceled", filesystem.HookDeleteTempFile)
-	//fs.Use("AfterUpload", filesystem.SlaveAfterUpload)
-	//fs.Use("AfterValidateFailed", filesystem.HookDeleteTempFile)
-	//
-	////// 是否允许覆盖
-	////if c.Request.Header.Get("X-Cr-Overwrite") == "false" {
-	////	fileData.Mode = fsctx.Create
-	////}
-	//
-	//// 执行上传
-	//err = fs.LocalUpload(ctx, &fileData)
-	//if err != nil {
-	//	c.JSON(200, serializer.Err(serializer.CodeUploadFailed, err.Error(), err))
-	//	return
-	//}
-	//
-	//c.JSON(200, serializer.Response{
-	//	Code: 0,
-	//})
 }
 
 // SlaveGetUploadSession 从机创建上传会话
@@ -116,6 +45,21 @@ func SlaveGetUploadSession(c *gin.Context) {
 	}
 }
 
+// SlaveDeleteUploadSession 从机删除上传会话
+func SlaveDeleteUploadSession(c *gin.Context) {
+	// 创建上下文
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var service explorer.UploadSessionService
+	if err := c.ShouldBindUri(&service); err == nil {
+		res := service.SlaveDelete(ctx, c)
+		c.JSON(200, res)
+	} else {
+		c.JSON(200, ErrorResponse(err))
+	}
+}
+
 // SlaveDownload 从机文件下载,此请求返回的HTTP状态码不全为200
 func SlaveDownload(c *gin.Context) {
 	// 创建上下文

routers/router.go

@@ -46,9 +46,15 @@ func InitSlaveRouter() *gin.Engine {
 		// 接收主机心跳包
 		v3.POST("heartbeat", controllers.SlaveHeartbeat)
 		// 上传
-		v3.POST("upload/:sessionId", controllers.SlaveUpload)
+		upload := v3.Group("upload")
+		{
+			// 上传分片
+			upload.POST(":sessionId", controllers.SlaveUpload)
 			// 创建上传会话上传
-		v3.PUT("upload", controllers.SlaveGetUploadSession)
+			upload.PUT("", controllers.SlaveGetUploadSession)
+			// 删除上传会话
+			upload.DELETE(":sessionId", controllers.SlaveDeleteUploadSession)
+		}
 		// 下载
 		v3.GET("download/:speed/:path/:name", controllers.SlaveDownload)
 		// 预览 / 外链

service/explorer/upload.go

@@ -241,6 +241,28 @@ func (service *UploadSessionService) Delete(ctx context.Context, c *gin.Context)
 	return serializer.Response{}
 }
 
+// SlaveDelete 从机删除指定上传会话
+func (service *UploadSessionService) SlaveDelete(ctx context.Context, c *gin.Context) serializer.Response {
+	// 创建文件系统
+	fs, err := filesystem.NewAnonymousFileSystem()
+	if err != nil {
+		return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err)
+	}
+	defer fs.Recycle()
+
+	session, ok := cache.Get(filesystem.UploadSessionCachePrefix + service.ID)
+	if !ok {
+		return serializer.Err(serializer.CodeUploadSessionExpired, "Slave Upload session file placeholder not exist", nil)
+	}
+
+	if _, err := fs.Handler.Delete(ctx, []string{session.(serializer.UploadSession).SavePath}); err != nil {
+		return serializer.Err(serializer.CodeInternalSetting, "Failed to delete temp file", err)
+	}
+
+	cache.Deletes([]string{service.ID}, filesystem.UploadSessionCachePrefix)
+	return serializer.Response{}
+}
+
 // DeleteAllUploadSession 删除当前用户的全部上传绘会话
 func DeleteAllUploadSession(ctx context.Context, c *gin.Context) serializer.Response {
 	// 创建文件系统