fix(115_open): refresh expired OSS credentials during upload

drivers/115_open/upload.go

@@ -3,6 +3,7 @@ package _115_open
 import (
 	"context"
 	"encoding/base64"
+	"errors"
 	"io"
 	"time"
 
@@ -70,6 +71,19 @@ func (d *Open115) singleUpload(ctx context.Context, tempF model.File, tokenResp
 // 	} `json:"data"`
 // }
 
+// retryExpiredToken retries only the rejected OSS operation, preserving the upload ID.
+func retryExpiredToken(refresh func() error, operation func() error) error {
+	err := operation()
+	var serviceErr oss.ServiceError
+	if !errors.As(err, &serviceErr) || serviceErr.Code != "SecurityTokenExpired" {
+		return err
+	}
+	if err := refresh(); err != nil {
+		return err
+	}
+	return operation()
+}
+
 func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer, up driver.UpdateProgress, tokenResp *sdk.UploadGetTokenResp, initResp *sdk.UploadInitResp) error {
 	ossClient, err := netutil.NewOSSClient(tokenResp.Endpoint, tokenResp.AccessKeyId, tokenResp.AccessKeySecret, oss.SecurityToken(tokenResp.SecurityToken))
 	if err != nil {
@@ -80,7 +94,32 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
 		return err
 	}
 
-	imur, err := bucket.InitiateMultipartUpload(initResp.Object, oss.Sequential())
+	refresh := func() error {
+		if err := d.WaitLimit(ctx); err != nil {
+			return err
+		}
+		token, err := d.client.UploadGetToken(ctx)
+		if err != nil {
+			return err
+		}
+		client, err := netutil.NewOSSClient(token.Endpoint, token.AccessKeyId, token.AccessKeySecret, oss.SecurityToken(token.SecurityToken))
+		if err != nil {
+			return err
+		}
+		newBucket, err := client.Bucket(initResp.Bucket)
+		if err != nil {
+			return err
+		}
+		bucket = newBucket
+		return nil
+	}
+
+	var imur oss.InitiateMultipartUploadResult
+	err = retryExpiredToken(refresh, func() error {
+		var err error
+		imur, err = bucket.InitiateMultipartUpload(initResp.Object, oss.Sequential(), oss.WithContext(ctx))
+		return err
+	})
 	if err != nil {
 		return err
 	}
@@ -109,13 +148,17 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
 			return err
 		}
 		err = retry.Do(func() error {
-			rd.Seek(0, io.SeekStart)
-			part, err := bucket.UploadPart(imur, driver.NewLimitedUploadStream(ctx, rd), partSize, int(i))
-			if err != nil {
-				return err
-			}
-			parts[i-1] = part
-			return nil
+			return retryExpiredToken(refresh, func() error {
+				if _, err := rd.Seek(0, io.SeekStart); err != nil {
+					return err
+				}
+				part, err := bucket.UploadPart(imur, driver.NewLimitedUploadStream(ctx, rd), partSize, int(i), oss.WithContext(ctx))
+				if err != nil {
+					return err
+				}
+				parts[i-1] = part
+				return nil
+			})
 		},
 			retry.Context(ctx),
 			retry.Attempts(3),
@@ -134,14 +177,16 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
 		up(float64(offset) * 100 / float64(fileSize))
 	}
 
-	// callbackRespBytes := make([]byte, 1024)
-	_, err = bucket.CompleteMultipartUpload(
-		imur,
-		parts,
-		oss.Callback(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.Callback))),
-		oss.CallbackVar(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.CallbackVar))),
-		// oss.CallbackResult(&callbackRespBytes),
-	)
+	err = retryExpiredToken(refresh, func() error {
+		_, err := bucket.CompleteMultipartUpload(
+			imur,
+			parts,
+			oss.Callback(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.Callback))),
+			oss.CallbackVar(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.CallbackVar))),
+			oss.WithContext(ctx),
+		)
+		return err
+	})
 	if err != nil {
 		return err
 	}