Skip to content

Commit f986a05

Browse files
xqmmcqscodex
andcommitted
fix(115_open): refresh expired OSS credentials during upload
- Refresh OSS credentials and retry the rejected multipart operation - Preserve upload IDs, completed parts, callbacks, and outer part retries - Rewind parts before resending and pass the upload context to OSS requests Co-authored-by: Codex <267193182+codex@users.noreply.github.com>
1 parent 6247cf7 commit f986a05

1 file changed

Lines changed: 61 additions & 16 deletions

File tree

drivers/115_open/upload.go

Lines changed: 61 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package _115_open
33
import (
44
"context"
55
"encoding/base64"
6+
"errors"
67
"io"
78
"time"
89

@@ -70,6 +71,19 @@ func (d *Open115) singleUpload(ctx context.Context, tempF model.File, tokenResp
7071
// } `json:"data"`
7172
// }
7273

74+
// retryExpiredToken retries only the rejected OSS operation, preserving the upload ID.
75+
func retryExpiredToken(refresh func() error, operation func() error) error {
76+
err := operation()
77+
var serviceErr oss.ServiceError
78+
if !errors.As(err, &serviceErr) || serviceErr.Code != "SecurityTokenExpired" {
79+
return err
80+
}
81+
if err := refresh(); err != nil {
82+
return err
83+
}
84+
return operation()
85+
}
86+
7387
func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer, up driver.UpdateProgress, tokenResp *sdk.UploadGetTokenResp, initResp *sdk.UploadInitResp) error {
7488
ossClient, err := netutil.NewOSSClient(tokenResp.Endpoint, tokenResp.AccessKeyId, tokenResp.AccessKeySecret, oss.SecurityToken(tokenResp.SecurityToken))
7589
if err != nil {
@@ -80,7 +94,32 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
8094
return err
8195
}
8296

83-
imur, err := bucket.InitiateMultipartUpload(initResp.Object, oss.Sequential())
97+
refresh := func() error {
98+
if err := d.WaitLimit(ctx); err != nil {
99+
return err
100+
}
101+
token, err := d.client.UploadGetToken(ctx)
102+
if err != nil {
103+
return err
104+
}
105+
client, err := netutil.NewOSSClient(token.Endpoint, token.AccessKeyId, token.AccessKeySecret, oss.SecurityToken(token.SecurityToken))
106+
if err != nil {
107+
return err
108+
}
109+
newBucket, err := client.Bucket(initResp.Bucket)
110+
if err != nil {
111+
return err
112+
}
113+
bucket = newBucket
114+
return nil
115+
}
116+
117+
var imur oss.InitiateMultipartUploadResult
118+
err = retryExpiredToken(refresh, func() error {
119+
var err error
120+
imur, err = bucket.InitiateMultipartUpload(initResp.Object, oss.Sequential(), oss.WithContext(ctx))
121+
return err
122+
})
84123
if err != nil {
85124
return err
86125
}
@@ -109,13 +148,17 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
109148
return err
110149
}
111150
err = retry.Do(func() error {
112-
rd.Seek(0, io.SeekStart)
113-
part, err := bucket.UploadPart(imur, driver.NewLimitedUploadStream(ctx, rd), partSize, int(i))
114-
if err != nil {
115-
return err
116-
}
117-
parts[i-1] = part
118-
return nil
151+
return retryExpiredToken(refresh, func() error {
152+
if _, err := rd.Seek(0, io.SeekStart); err != nil {
153+
return err
154+
}
155+
part, err := bucket.UploadPart(imur, driver.NewLimitedUploadStream(ctx, rd), partSize, int(i), oss.WithContext(ctx))
156+
if err != nil {
157+
return err
158+
}
159+
parts[i-1] = part
160+
return nil
161+
})
119162
},
120163
retry.Context(ctx),
121164
retry.Attempts(3),
@@ -134,14 +177,16 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
134177
up(float64(offset) * 100 / float64(fileSize))
135178
}
136179

137-
// callbackRespBytes := make([]byte, 1024)
138-
_, err = bucket.CompleteMultipartUpload(
139-
imur,
140-
parts,
141-
oss.Callback(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.Callback))),
142-
oss.CallbackVar(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.CallbackVar))),
143-
// oss.CallbackResult(&callbackRespBytes),
144-
)
180+
err = retryExpiredToken(refresh, func() error {
181+
_, err := bucket.CompleteMultipartUpload(
182+
imur,
183+
parts,
184+
oss.Callback(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.Callback))),
185+
oss.CallbackVar(base64.StdEncoding.EncodeToString([]byte(initResp.Callback.Value.CallbackVar))),
186+
oss.WithContext(ctx),
187+
)
188+
return err
189+
})
145190
if err != nil {
146191
return err
147192
}

0 commit comments

Comments
 (0)