dropbox: fix context cancellation (e.g. --max-duration limit) not stopping in-flight requests
This switches the calls to the dropbox SDK over to use the context taking variants. This allows context cancellation to propagate properly. Fixes #9688
This commit is contained in:
@@ -19,7 +19,7 @@ func (f *Fs) finishBatch(ctx context.Context, items []*files.UploadSessionFinish
|
||||
Entries: items,
|
||||
}
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
complete, err = f.srv.UploadSessionFinishBatchV2(arg)
|
||||
complete, err = f.srv.UploadSessionFinishBatchV2Context(ctx, arg)
|
||||
if retry, err := shouldRetryExclude(ctx, err); !retry {
|
||||
return retry, err
|
||||
}
|
||||
|
||||
+42
-42
@@ -387,11 +387,11 @@ type Fs struct {
|
||||
opt Options // parsed options
|
||||
ci *fs.ConfigInfo // global config
|
||||
features *fs.Features // optional features
|
||||
srv files.Client // the connection to the dropbox server
|
||||
svc files.Client // the connection to the dropbox server (unauthorized)
|
||||
sharing sharing.Client // as above, but for generating sharing links
|
||||
users users.Client // as above, but for accessing user information
|
||||
team team.Client // for the Teams API
|
||||
srv files.ContextClient // the connection to the dropbox server
|
||||
svc files.ContextClient // the connection to the dropbox server (unauthorized)
|
||||
sharing sharing.ContextClient // as above, but for generating sharing links
|
||||
users users.ContextClient // as above, but for accessing user information
|
||||
team team.ContextClient // for the Teams API
|
||||
slashRoot string // root with "/" prefix, lowercase
|
||||
slashRootSlash string // root with "/" prefix and postfix, lowercase
|
||||
pacer *fs.Pacer // To pace the API calls
|
||||
@@ -577,7 +577,7 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e
|
||||
}
|
||||
|
||||
// NOTE: needs to be created pre-impersonation so we can look up the impersonated user
|
||||
f.team = team.New(cfg)
|
||||
f.team = team.NewContext(cfg)
|
||||
|
||||
if opt.Impersonate != "" {
|
||||
user := team.UserSelectorArg{
|
||||
@@ -588,7 +588,7 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e
|
||||
members := []*team.UserSelectorArg{&user}
|
||||
args := team.NewMembersGetInfoArgs(members)
|
||||
|
||||
memberIDs, err := f.team.MembersGetInfo(args)
|
||||
memberIDs, err := f.team.MembersGetInfoContext(ctx, args)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid dropbox team member: %q: %w", opt.Impersonate, err)
|
||||
}
|
||||
@@ -603,10 +603,10 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e
|
||||
cfg.AsAdminID = opt.ImpersonateAdmin
|
||||
}
|
||||
|
||||
f.srv = files.New(cfg)
|
||||
f.svc = files.New(ucfg)
|
||||
f.sharing = sharing.New(cfg)
|
||||
f.users = users.New(cfg)
|
||||
f.srv = files.NewContext(cfg)
|
||||
f.svc = files.NewContext(ucfg)
|
||||
f.sharing = sharing.NewContext(cfg)
|
||||
f.users = users.NewContext(cfg)
|
||||
f.features = (&fs.Features{
|
||||
CaseInsensitive: true,
|
||||
ReadMimeType: false,
|
||||
@@ -673,7 +673,7 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e
|
||||
// If root starts with / then use the actual root
|
||||
var acc *users.FullAccount
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
acc, err = f.users.GetCurrentAccount()
|
||||
acc, err = f.users.GetCurrentAccountContext(ctx)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -737,7 +737,7 @@ type getMetadataResult struct {
|
||||
// getMetadata gets the metadata for a file or directory
|
||||
func (f *Fs) getMetadata(ctx context.Context, objPath string) (res getMetadataResult) {
|
||||
res.err = f.pacer.Call(func() (bool, error) {
|
||||
res.entry, res.err = f.srv.GetMetadata(&files.GetMetadataArg{
|
||||
res.entry, res.err = f.srv.GetMetadataContext(ctx, &files.GetMetadataArg{
|
||||
Path: f.opt.Enc.FromStandardPath(objPath),
|
||||
})
|
||||
return shouldRetry(ctx, res.err)
|
||||
@@ -912,7 +912,7 @@ func (f *Fs) listSharedFolders(ctx context.Context, callback func(fs.DirEntry) e
|
||||
Limit: 100,
|
||||
}
|
||||
err := f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.sharing.ListFolders(&arg)
|
||||
res, err = f.sharing.ListFoldersContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -924,7 +924,7 @@ func (f *Fs) listSharedFolders(ctx context.Context, callback func(fs.DirEntry) e
|
||||
Cursor: res.Cursor,
|
||||
}
|
||||
err := f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.sharing.ListFoldersContinue(&arg)
|
||||
res, err = f.sharing.ListFoldersContinueContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -973,7 +973,7 @@ func (f *Fs) mountSharedFolder(ctx context.Context, id string) error {
|
||||
SharedFolderId: id,
|
||||
}
|
||||
err := f.pacer.Call(func() (bool, error) {
|
||||
_, err := f.sharing.MountFolder(&arg)
|
||||
_, err := f.sharing.MountFolderContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
return err
|
||||
@@ -990,7 +990,7 @@ func (f *Fs) listReceivedFiles(ctx context.Context, callback func(fs.DirEntry) e
|
||||
Limit: 100,
|
||||
}
|
||||
err := f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.sharing.ListReceivedFiles(&arg)
|
||||
res, err = f.sharing.ListReceivedFilesContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1002,7 +1002,7 @@ func (f *Fs) listReceivedFiles(ctx context.Context, callback func(fs.DirEntry) e
|
||||
Cursor: res.Cursor,
|
||||
}
|
||||
err := f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.sharing.ListReceivedFilesContinue(&arg)
|
||||
res, err = f.sharing.ListReceivedFilesContinueContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1109,7 +1109,7 @@ func (f *Fs) ListP(ctx context.Context, dir string, callback fs.ListRCallback) (
|
||||
arg.Path = "" // Specify root folder as empty string
|
||||
}
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.srv.ListFolder(arg)
|
||||
res, err = f.srv.ListFolderContext(ctx, arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1127,7 +1127,7 @@ func (f *Fs) ListP(ctx context.Context, dir string, callback fs.ListRCallback) (
|
||||
Cursor: res.Cursor,
|
||||
}
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.srv.ListFolderContinue(&arg)
|
||||
res, err = f.srv.ListFolderContinueContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1164,7 +1164,7 @@ func (f *Fs) ListP(ctx context.Context, dir string, callback fs.ListRCallback) (
|
||||
var sfMeta *sharing.SharedFolderMetadata
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
var apiErr error
|
||||
sfMeta, apiErr = f.sharing.GetFolderMetadata(sharing.NewGetMetadataArgs(folderInfo.SharingInfo.SharedFolderId))
|
||||
sfMeta, apiErr = f.sharing.GetFolderMetadataContext(ctx, sharing.NewGetMetadataArgs(folderInfo.SharingInfo.SharedFolderId))
|
||||
return shouldRetry(ctx, apiErr)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1251,7 +1251,7 @@ func (f *Fs) Mkdir(ctx context.Context, dir string) error {
|
||||
return cErr
|
||||
}
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
_, err = f.srv.CreateFolderV2(&arg2)
|
||||
_, err = f.srv.CreateFolderV2Context(ctx, &arg2)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
return err
|
||||
@@ -1279,7 +1279,7 @@ func (f *Fs) purgeCheck(ctx context.Context, dir string, check bool) (err error)
|
||||
}
|
||||
var res *files.ListFolderResult
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
res, err = f.srv.ListFolder(arg)
|
||||
res, err = f.srv.ListFolderContext(ctx, arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1303,7 +1303,7 @@ func (f *Fs) purgeCheck(ctx context.Context, dir string, check bool) (err error)
|
||||
|
||||
// remove it
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
_, err = f.srv.DeleteV2(&files.DeleteArg{Path: encRoot})
|
||||
_, err = f.srv.DeleteV2Context(ctx, &files.DeleteArg{Path: encRoot})
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
return err
|
||||
@@ -1362,7 +1362,7 @@ func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (dst fs.Obj
|
||||
}
|
||||
var result *files.RelocationResult
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
result, err = f.srv.CopyV2(&arg)
|
||||
result, err = f.srv.CopyV2Context(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1423,7 +1423,7 @@ func (f *Fs) Move(ctx context.Context, src fs.Object, remote string) (fs.Object,
|
||||
var err error
|
||||
var result *files.RelocationResult
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
result, err = f.srv.MoveV2(&arg)
|
||||
result, err = f.srv.MoveV2Context(ctx, &arg)
|
||||
switch e := err.(type) {
|
||||
case files.MoveV2APIError:
|
||||
// There seems to be a bit of eventual consistency here which causes this to
|
||||
@@ -1477,7 +1477,7 @@ func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration,
|
||||
|
||||
var linkRes sharing.IsSharedLinkMetadata
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
linkRes, err = f.sharing.CreateSharedLinkWithSettings(&createArg)
|
||||
linkRes, err = f.sharing.CreateSharedLinkWithSettingsContext(ctx, &createArg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
|
||||
@@ -1486,7 +1486,7 @@ func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration,
|
||||
fs.Debugf(absPath, "can't create link with expiry, trying without")
|
||||
createArg.Settings.Expires = nil
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
linkRes, err = f.sharing.CreateSharedLinkWithSettings(&createArg)
|
||||
linkRes, err = f.sharing.CreateSharedLinkWithSettingsContext(ctx, &createArg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
}
|
||||
@@ -1500,7 +1500,7 @@ func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration,
|
||||
}
|
||||
var listRes *sharing.ListSharedLinksResult
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
listRes, err = f.sharing.ListSharedLinks(&listArg)
|
||||
listRes, err = f.sharing.ListSharedLinksContext(ctx, &listArg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1561,7 +1561,7 @@ func (f *Fs) DirMove(ctx context.Context, src fs.Fs, srcRemote, dstRemote string
|
||||
},
|
||||
}
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
_, err = f.srv.MoveV2(&arg)
|
||||
_, err = f.srv.MoveV2Context(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1575,7 +1575,7 @@ func (f *Fs) DirMove(ctx context.Context, src fs.Fs, srcRemote, dstRemote string
|
||||
func (f *Fs) About(ctx context.Context) (usage *fs.Usage, err error) {
|
||||
var q *users.SpaceUsage
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
q, err = f.users.GetSpaceUsage()
|
||||
q, err = f.users.GetSpaceUsageContext(ctx)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1662,7 +1662,7 @@ func (f *Fs) changeNotifyCursor(ctx context.Context) (cursor string, err error)
|
||||
arg.Path = ""
|
||||
}
|
||||
|
||||
startCursor, err = f.srv.ListFolderGetLatestCursor(arg)
|
||||
startCursor, err = f.srv.ListFolderGetLatestCursorContext(ctx, arg)
|
||||
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
@@ -1695,7 +1695,7 @@ func (f *Fs) changeNotifyRunner(ctx context.Context, notifyFunc func(string, fs.
|
||||
Timeout: timeout,
|
||||
}
|
||||
|
||||
res, err = f.svc.ListFolderLongpoll(&args)
|
||||
res, err = f.svc.ListFolderLongpollContext(ctx, &args)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1718,7 +1718,7 @@ func (f *Fs) changeNotifyRunner(ctx context.Context, notifyFunc func(string, fs.
|
||||
Cursor: cursor,
|
||||
}
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
changeList, err = f.srv.ListFolderContinue(&arg)
|
||||
changeList, err = f.srv.ListFolderContinueContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1969,7 +1969,7 @@ func (o *Object) export(ctx context.Context) (in io.ReadCloser, err error) {
|
||||
arg := files.ExportArg{Path: o.id, ExportFormat: string(o.exportAPIFormat)}
|
||||
var exportResult *files.ExportResult
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
exportResult, in, err = o.fs.srv.Export(&arg)
|
||||
exportResult, in, err = o.fs.srv.ExportContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -1991,7 +1991,7 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.Read
|
||||
Url: o.url,
|
||||
}
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
_, in, err = o.fs.sharing.GetSharedLinkFile(&arg)
|
||||
_, in, err = o.fs.sharing.GetSharedLinkFileContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -2011,7 +2011,7 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.Read
|
||||
ExtraHeaders: headers,
|
||||
}
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
_, in, err = o.fs.srv.Download(&arg)
|
||||
_, in, err = o.fs.srv.DownloadContext(ctx, &arg)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
|
||||
@@ -2035,7 +2035,7 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
|
||||
// start upload
|
||||
var res *files.UploadSessionStartResult
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
res, err = o.fs.srv.UploadSessionStart(&files.UploadSessionStartArg{}, nil)
|
||||
res, err = o.fs.srv.UploadSessionStartContext(ctx, &files.UploadSessionStartArg{}, nil)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
if err != nil {
|
||||
@@ -2081,7 +2081,7 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
|
||||
if _, err = chunk.Seek(skip, io.SeekStart); err != nil {
|
||||
return false, err
|
||||
}
|
||||
err = o.fs.srv.UploadSessionAppendV2(&appendArg, chunk)
|
||||
err = o.fs.srv.UploadSessionAppendV2Context(ctx, &appendArg, chunk)
|
||||
// after session is started, we retry everything
|
||||
if err != nil {
|
||||
// Check for incorrect offset error and retry with new offset
|
||||
@@ -2140,7 +2140,7 @@ func (o *Object) uploadChunked(ctx context.Context, in0 io.Reader, commitInfo *f
|
||||
}
|
||||
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
entry, err = o.fs.srv.UploadSessionFinish(args, nil)
|
||||
entry, err = o.fs.srv.UploadSessionFinishContext(ctx, args, nil)
|
||||
if retry, err := shouldRetryExclude(ctx, err); !retry {
|
||||
return retry, err
|
||||
}
|
||||
@@ -2208,7 +2208,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
|
||||
entry, err = o.uploadChunked(ctx, in, commitInfo, size)
|
||||
} else {
|
||||
err = o.fs.pacer.CallNoRetry(func() (bool, error) {
|
||||
entry, err = o.fs.srv.Upload(&files.UploadArg{CommitInfo: *commitInfo}, in)
|
||||
entry, err = o.fs.srv.UploadContext(ctx, &files.UploadArg{CommitInfo: *commitInfo}, in)
|
||||
return shouldRetry(ctx, err)
|
||||
})
|
||||
}
|
||||
@@ -2233,7 +2233,7 @@ func (o *Object) Remove(ctx context.Context) (err error) {
|
||||
return errNotSupportedInSharedMode
|
||||
}
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
_, err = o.fs.srv.DeleteV2(&files.DeleteArg{
|
||||
_, err = o.fs.srv.DeleteV2Context(ctx, &files.DeleteArg{
|
||||
Path: o.fs.opt.Enc.FromStandardPath(o.remotePath()),
|
||||
})
|
||||
return shouldRetry(ctx, err)
|
||||
|
||||
@@ -3,8 +3,11 @@ package dropbox
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/dropbox/dropbox-sdk-go-unofficial/v6/dropbox"
|
||||
"github.com/dropbox/dropbox-sdk-go-unofficial/v6/dropbox/files"
|
||||
@@ -16,11 +19,11 @@ import (
|
||||
)
|
||||
|
||||
type paperMetadataClient struct {
|
||||
files.Client
|
||||
files.ContextClient
|
||||
info *files.FileMetadata
|
||||
}
|
||||
|
||||
func (c paperMetadataClient) GetMetadata(arg *files.GetMetadataArg) (files.IsMetadata, error) {
|
||||
func (c paperMetadataClient) GetMetadataContext(ctx context.Context, arg *files.GetMetadataArg) (files.IsMetadata, error) {
|
||||
if arg.Path == "document" {
|
||||
return c.info, nil
|
||||
}
|
||||
@@ -35,6 +38,48 @@ func (c paperMetadataClient) GetMetadata(arg *files.GetMetadataArg) (files.IsMet
|
||||
}
|
||||
}
|
||||
|
||||
func TestInternalGetMetadataCancellation(t *testing.T) {
|
||||
requestStarted := make(chan struct{})
|
||||
releaseRequest := make(chan struct{})
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
close(requestStarted)
|
||||
<-releaseRequest
|
||||
}))
|
||||
defer server.Close()
|
||||
defer close(releaseRequest)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
f := &Fs{
|
||||
srv: files.NewContext(dropbox.Config{
|
||||
Client: server.Client(),
|
||||
URLGenerator: func(hostType string, namespace string, route string) string {
|
||||
return server.URL
|
||||
},
|
||||
}),
|
||||
pacer: fs.NewPacer(ctx, pacer.NewDefault(pacer.MinSleep(0), pacer.MaxSleep(time.Millisecond))),
|
||||
}
|
||||
|
||||
result := make(chan getMetadataResult, 1)
|
||||
go func() {
|
||||
result <- f.getMetadata(ctx, "/file")
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-requestStarted:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Dropbox request did not start")
|
||||
}
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case res := <-result:
|
||||
require.ErrorIs(t, res.err, context.Canceled)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Dropbox request did not observe cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestInternalCheckPathLength(t *testing.T) {
|
||||
rep := func(n int, r rune) (out string) {
|
||||
rs := make([]rune, n)
|
||||
@@ -110,7 +155,7 @@ Lorem ipsum __dolor__ sit amet
|
||||
var err error
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
reader := strings.NewReader(content)
|
||||
_, err = f.srv.PaperCreate(&arg, reader)
|
||||
_, err = f.srv.PaperCreateContext(context.Background(), &arg, reader)
|
||||
return shouldRetry(context.Background(), err)
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
Reference in New Issue
Block a user